Merge pull request #1332 from chaitin/ci-fix-fifth-scan-findings

fix(management,mcp-server): attribute control channel answers and bou…

Author: jlongWang

management/tcontrollerd/controller/website.go

@@ -12,6 +12,7 @@ import (
 	"path/filepath"
 	"strings"
 
+	"chaitin.cn/dev/go/errors"
 	"chaitin.cn/patronus/safeline-2/management/tcontrollerd/model"
 	"chaitin.cn/patronus/safeline-2/management/tcontrollerd/pkg/ngcmd"
 	pb "chaitin.cn/patronus/safeline-2/management/tcontrollerd/proto/website"
@@ -132,6 +133,16 @@ func generateFullConfigAndReload(msg []byte) error {
 		return err
 	}
 
+	// "null" is what a query that returned nothing looks like on the wire, and
+	// a full push that carries it would delete every configuration of the
+	// installation. The management server always sends an array, so a payload
+	// that is not a list is refused and the current configuration is kept: an
+	// installation that really has no site sends an empty array, which is
+	// applied.
+	if websites == nil {
+		return errors.New("the full push does not carry a list of sites, keeping the current configuration")
+	}
+
 	if err := utils.EnsureDir(nginxConfigPath); err != nil {
 		return err
 	}

management/webserver/pkg/fvm/fsl/quote.go

@@ -28,7 +28,11 @@ func appendEscapedByte(buf []byte, b byte) []byte {
 	case '\\':
 		return append(buf, `\\`...)
 	default:
-		return append(buf, fmt.Sprintf("\\x%x", b)...)
+		// The escape has to be two digits wide: "\x5" followed by a literal
+		// "a" is read by the lexer as "\x5a" ("Z"), so a rule whose comment or
+		// header value ends with a control byte would silently change the
+		// source that is handed to the compiler.
+		return append(buf, fmt.Sprintf("\\x%02x", b)...)
 	}
 }
 

management/webserver/rpc/website.go

@@ -31,6 +31,20 @@ func Publish(msg []byte, eventType string) error {
 	if sub == nil {
 		return errors.New("Service is abnormal, and the nginx conf cannot be updated for the time being. Please go to the shell to check the relevant logs")
 	}
+
+	// Only one exchange may be outstanding on the stream. The response carries
+	// no request id, so two overlapping callers that share the single slot below
+	// would each read the answer of the other: one would commit a database
+	// change that tcontrollerd refused, the other would roll back one it
+	// applied. The generated client does not allow concurrent sends on one
+	// stream either.
+	sub.exchangeMu.Lock()
+	defer sub.exchangeMu.Unlock()
+
+	// A response whose own request already gave up is still in the slot, and
+	// this call would read it as its own answer.
+	discardPendingResponses(sub)
+
 	err := sub.stream.Send(&pb.Event{
 		Type: eventType,
 		Msg:  msg,
@@ -59,18 +73,49 @@ func Publish(msg []byte, eventType string) error {
 				return nil
 			}
 		case <-rspTimeoutTicker.C:
+			// The device did not answer, so what it applied is unknown: an
+			// answer that arrives later would be read by the next request as
+			// if it belonged to it. The stream is dropped instead, which
+			// leaves the slot this call abandons unreachable and makes
+			// tcontrollerd reconnect and synchronise with a full push.
+			clearSubscriber(sub)
+			select {
+			case sub.errCh <- errors.New("Wait timeout for updating result"):
+			default:
+			}
+
 			return errors.New("Wait timeout for updating result")
 		}
 	}
 }
 
+// discardPendingResponses drops the responses that are sitting in the slot.
+//
+// It is called while the exchange of the caller is held, so nothing in the slot
+// can belong to a request that is still waiting.
+func discardPendingResponses(sc *StreamClient) {
+	for {
+		select {
+		case rsp := <-sc.rspCh:
+			logger.Warnf("Discarding an unclaimed control channel response of type %s", rsp.GetType())
+		default:
+			return
+		}
+	}
+}
+
 // StreamClient is the instance for every client stream connected
 type StreamClient struct {
 	stream pb.Website_SubscribeServer
 	timer  *time.Timer // timer for timeout
 	errCh  chan error
 	rspCh  chan *pb.Response
 	quit   chan struct{} // quit stream client gracefully
+
+	// exchangeMu serialises the send-and-wait of Publish. The response of a
+	// request is not labelled with the request it answers, so at most one
+	// request may be in flight at a time.
+	exchangeMu sync.Mutex
 }
 
 // currentSubscriber returns the active subscriber, or nil when no stream is
@@ -171,7 +216,18 @@ func (sc *StreamClient) recvLoop() {
 		if rsp.Type != Pong {
 			// receive response
 			logger.Infof("Recv updating website rsp: err(%t), msg(%s)", rsp.GetErr(), rsp.GetMsg())
-			sc.rspCh <- rsp
+
+			// This loop is also the goroutine that reads the keepalive pongs
+			// and resets the timer, so it must not block on a full slot: it
+			// would disconnect a stream whose peer is perfectly healthy. Only
+			// one exchange is outstanding at a time (see Publish) and it
+			// empties the slot before it sends, so a slot that is already full
+			// holds a response nobody is waiting for.
+			select {
+			case sc.rspCh <- rsp:
+			default:
+				logger.Warn("Discarding a control channel response that no request is waiting for")
+			}
 			continue
 		}
 		sc.timer.Reset(KeepaliveTimeout)
@@ -195,7 +251,21 @@ func GetWebsiteServer() *WebsiteServer {
 func publishFullWebsite() error {
 	var websites []model.Website
 	db := database.GetDB().DB
-	db.Model(&model.Website{}).Find(&websites)
+
+	// A failed query leaves the slice empty, and an empty list means "this
+	// installation has no sites" to tcontrollerd: it would delete every nginx
+	// configuration it manages. The error has to stop the push instead.
+	if err := db.Model(&model.Website{}).Find(&websites).Error; err != nil {
+		return errors.Wrap(err, "load websites")
+	}
+
+	if websites == nil {
+		// A nil slice marshals to "null", which tcontrollerd reads as a
+		// malformed push. An installation that really has no site sends an
+		// empty list.
+		websites = []model.Website{}
+	}
+
 	byteWebsites, err := json.Marshal(&websites)
 	if err != nil {
 		return err

management/webserver/utils/httpclient.go

@@ -1,6 +1,7 @@
 package utils
 
 import (
+	"net"
 	"net/http"
 	"net/url"
 	"os"
@@ -10,6 +11,21 @@ import (
 
 const proxyName = "HTTPS_PROXY"
 
+const (
+	// httpClientTimeout bounds one request end to end, from the connection
+	// attempt to the last byte of the body.
+	httpClientTimeout = 15 * time.Second
+
+	// httpClientDialTimeout bounds the connection attempt, and
+	// httpClientTLSHandshakeTimeout the handshake that follows it.
+	httpClientDialTimeout         = 5 * time.Second
+	httpClientTLSHandshakeTimeout = 5 * time.Second
+
+	// httpClientResponseHeaderTimeout bounds the wait for the response header
+	// once the request was written.
+	httpClientResponseHeaderTimeout = 10 * time.Second
+)
+
 var (
 	httpClientOnce sync.Once
 	httpClient     *http.Client
@@ -20,14 +36,27 @@ var (
 // The client is built once: the API handlers and the cron jobs call this at the
 // same time, and the unsynchronized lazy assignment used before let two of them
 // build (and see) a client at the same time.
+//
+// Every phase of a request is bounded. The client is used by endpoints that any
+// caller can reach, such as the upgrade tip lookup, and the endpoints it talks
+// to are remote: an upstream that accepts the connection and then stops
+// answering would otherwise hold the handler goroutine and its socket for as
+// long as the connection lives, which a caller can repeat without limit.
 func GetHTTPClient() *http.Client {
 	httpClientOnce.Do(func() {
 		// Certificates are verified: this client talks to the telemetry
 		// endpoint and to the upgrade server, both of which are remote and can
 		// be spoofed by a man in the middle when verification is disabled.
 		tr := &http.Transport{
-			MaxIdleConns:    10,
-			IdleConnTimeout: 30 * time.Second,
+			DialContext: (&net.Dialer{
+				Timeout:   httpClientDialTimeout,
+				KeepAlive: 30 * time.Second,
+			}).DialContext,
+			MaxIdleConns:          10,
+			IdleConnTimeout:       30 * time.Second,
+			TLSHandshakeTimeout:   httpClientTLSHandshakeTimeout,
+			ResponseHeaderTimeout: httpClientResponseHeaderTimeout,
+			ExpectContinueTimeout: 1 * time.Second,
 		}
 
 		proxyUrl, existed := os.LookupEnv(proxyName)
@@ -36,7 +65,7 @@ func GetHTTPClient() *http.Client {
 			tr.Proxy = http.ProxyURL(uri)
 		}
 
-		httpClient = &http.Client{Transport: tr}
+		httpClient = &http.Client{Transport: tr, Timeout: httpClientTimeout}
 	})
 
 	return httpClient

mcp_server/README.md

@@ -44,6 +44,7 @@ instances:
     token_file: "/run/secrets/production-a.token"
     timeout: 30
     debug: false
+    ca_file: ""
     insecure_skip_verify: false
 
   - id: "production-b"
@@ -52,15 +53,23 @@ instances:
     token_file: "/run/secrets/production-b.token"
     timeout: 30
     debug: false
+    ca_file: ""
     insecure_skip_verify: false
 ```
 
 `LISTEN_ADDRESS` and `LISTEN_PORT` override listener settings. There is deliberately no global `SAFELINE_API_TOKEN`: every instance uses a separate token file.
 
+`ca_file` is the certificate authority that signed the certificate of the
+instance. A SafeLine console serves the certificate it generated for itself,
+which no public authority signed: point `ca_file` at that certificate (or at the
+authority that issued it) and the connection stays verified. Relative paths are
+resolved from the directory of the configuration file.
+
 `insecure_skip_verify` disables TLS verification for the SafeLine API. The API
 Token is an administrator credential that is sent on every request, so leave the
-setting at `false` and give the instance a certificate the server trusts. Enable
-it only for a lab instance whose certificate cannot be replaced.
+setting at `false` and use `ca_file` for an instance with a self-signed
+certificate. Setting it to `true` logs a warning at startup, and it cannot be
+combined with `ca_file`.
 
 Each `display_name` must be unique, ignoring letter case, and must not match another instance's `id`. During server discovery (or legacy initialization), the server publishes only the `display_name` to `instance_id` mappings as Server Instructions. This lets a user refer to a friendly name while the AI still calls `get_attack_events` with the stable, explicit `instance_id`. Instance addresses and credentials are never included in those instructions.
 

mcp_server/config.yaml

@@ -21,6 +21,11 @@ instances:
     token_file: "/run/secrets/safeline-a.token"
     timeout: 30
     debug: false
+    # ca_file is the certificate authority that signed the certificate of the
+    # instance. A SafeLine console serves the certificate it generated for
+    # itself, which no public authority signed: point this at that certificate
+    # (or at its CA) instead of switching the check off.
+    ca_file: ""
     # Leave TLS verification enabled: the SafeLine API Token is an
     # administrator credential and disabling verification lets anyone on the
     # path read it. Set this to true only for a lab instance with an
@@ -33,4 +38,5 @@ instances:
     token_file: "/run/secrets/safeline-b.token"
     timeout: 30
     debug: false
+    ca_file: ""
     insecure_skip_verify: false

mcp_server/internal/api/client.go

@@ -3,11 +3,13 @@ package api
 import (
 	"context"
 	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"
 
@@ -26,6 +28,25 @@ const (
 	maxRedirects = 10
 )
 
+// loadRootCAs reads the certificate authority the instance is issued by.
+//
+// It is a file rather than the system pool because the installation that the
+// MCP server talks to usually carries the certificate it generated for itself,
+// which no system pool knows about.
+func loadRootCAs(path string) (*x509.CertPool, error) {
+	content, err := os.ReadFile(path)
+	if err != nil {
+		return nil, errors.Wrap(err, "read ca_file failed")
+	}
+
+	roots := x509.NewCertPool()
+	if !roots.AppendCertsFromPEM(content) {
+		return nil, errors.New("ca_file does not contain a PEM encoded certificate")
+	}
+
+	return roots, nil
+}
+
 // Client is the transport used by a single SafeLine instance. Its request
 // method is intentionally unexported; APIClient exposes only approved read-only
 // operations.
@@ -35,7 +56,7 @@ type Client struct {
 	headers    http.Header
 }
 
-func newClient(baseURL string, timeout time.Duration, insecureSkipVerify bool, token string) (*Client, error) {
+func newClient(baseURL string, timeout time.Duration, insecureSkipVerify bool, caFile, token string) (*Client, error) {
 	parsedURL, err := url.Parse(strings.TrimSpace(baseURL))
 	if err != nil {
 		return nil, errors.Wrap(err, "parse base_url failed")
@@ -51,6 +72,27 @@ func newClient(baseURL string, timeout time.Duration, insecureSkipVerify bool, t
 	transport := &http.Transport{
 		TLSClientConfig: &tls.Config{InsecureSkipVerify: insecureSkipVerify}, // #nosec G402 -- deployment-controlled compatibility setting
 	}
+
+	if caFile != "" {
+		// The console of a SafeLine installation serves a certificate that no
+		// public authority signed, so the choice used to be between switching
+		// verification off and failing every request. Pinning the authority of
+		// the instance keeps it on, which is what keeps the deployment token in
+		// the request header out of the hands of anything on the path.
+		roots, err := loadRootCAs(caFile)
+		if err != nil {
+			return nil, err
+		}
+		transport.TLSClientConfig.RootCAs = roots
+	}
+
+	if insecureSkipVerify {
+		logger.With("base_url", parsedURL.Redacted()).Warn(
+			"insecure_skip_verify is set: the certificate of this instance is not checked, " +
+				"so anything on the path can read the API token of the deployment. " +
+				"Set ca_file to the certificate authority of the instance instead.")
+	}
+
 	client := &Client{
 		baseURL: parsedURL,
 		httpClient: &http.Client{

mcp_server/internal/api/client_redirect_test.go

@@ -29,7 +29,7 @@ func TestClientRefusesCrossOriginRedirect(t *testing.T) {
 	}))
 	defer origin.Close()
 
-	client, err := newClient(origin.URL, 5*time.Second, false, "deployment-token")
+	client, err := newClient(origin.URL, 5*time.Second, false, "", "deployment-token")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -61,7 +61,7 @@ func TestClientFollowsSameOriginRedirect(t *testing.T) {
 	server := httptest.NewServer(mux)
 	defer server.Close()
 
-	client, err := newClient(server.URL, 5*time.Second, false, "deployment-token")
+	client, err := newClient(server.URL, 5*time.Second, false, "", "deployment-token")
 	if err != nil {
 		t.Fatal(err)
 	}