Files
wehub-resource-sync 5357c39144
Fuzzer / Run Fuzzer (push) Has been cancelled
Race tests / Go race tests (ubuntu-22.04) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:01:40 +08:00

570 lines
22 KiB
Go

// Copyright 2019 Dolthub, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package servercfg
import (
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"net"
"os"
"path/filepath"
"runtime"
"strings"
"time"
"github.com/dolthub/go-mysql-server/sql"
)
var DefaultUnixSocketFilePath = DefaultMySQLUnixSocketFilePath
// LogLevel defines the available levels of logging for the server.
type LogLevel string
const (
LogLevel_Trace LogLevel = "trace"
LogLevel_Debug LogLevel = "debug"
LogLevel_Info LogLevel = "info"
LogLevel_Warning LogLevel = "warning"
LogLevel_Error LogLevel = "error"
LogLevel_Fatal LogLevel = "fatal"
LogLevel_Panic LogLevel = "panic"
)
type LogFormat string
const (
LogFormat_Text LogFormat = "text"
LogFormat_JSON LogFormat = "json"
)
const (
DefaultHost = "localhost"
DefaultPort = 3306
DefaultUser = "root"
DefaultPass = ""
DefaultTimeout = 8 * 60 * 60 * 1000 // 8 hours, same as MySQL
DefaultReadOnly = false
DefaultLogLevel = LogLevel_Info
DefaultLogFormat = LogFormat_Text
DefaultAutoCommit = true
DefaultAutoGCBehaviorEnable = true
DefaultDoltTransactionCommit = false
DefaultBranchActivityTracking = false
DefaultMaxConnections = 1000
DefaultMaxWaitConnections = 50
DefaultMaxWaitConnectionsTimeout = 60 * time.Second
DefaultDataDir = "."
DefaultCfgDir = ".doltcfg"
DefaultPrivilegeFilePath = "privileges.db"
DefaultBranchControlFilePath = "branch_control.db"
DefaultMetricsHost = ""
DefaultMetricsPort = -1
DefaultMCPPort = 7007
DefaultAllowCleartextPasswords = false
DefaultMySQLUnixSocketFilePath = "/tmp/mysql.sock"
DefaultMaxLoggedQueryLen = 0
DefaultEncodeLoggedQuery = false
DefaultCompressionLevel = 1
)
func ptr[T any](t T) *T {
return &t
}
// String returns the string representation of the log level.
func (level LogLevel) String() string {
switch level {
case LogLevel_Trace:
fallthrough
case LogLevel_Debug:
fallthrough
case LogLevel_Info:
fallthrough
case LogLevel_Warning:
fallthrough
case LogLevel_Error:
fallthrough
case LogLevel_Fatal:
fallthrough
case LogLevel_Panic:
return string(level)
default:
return "unknown"
}
}
type ClusterConfig interface {
StandbyRemotes() []ClusterStandbyRemoteConfig
BootstrapRole() string
BootstrapEpoch() int
RemotesAPIConfig() ClusterRemotesAPIConfig
}
type ClusterRemotesAPIConfig interface {
Address() string
Port() int
TLSKey() string
TLSCert() string
TLSCA() string
ServerNameURLMatches() []string
ServerNameDNSMatches() []string
}
type ClusterStandbyRemoteConfig interface {
Name() string
RemoteURLTemplate() string
}
type JwksConfig struct {
Name string `yaml:"name"`
LocationUrl string `yaml:"location_url"`
Claims map[string]string `yaml:"claims"`
FieldsToLog []string `yaml:"fields_to_log"`
}
// ServerConfig contains all of the configurable options for the MySQL-compatible server.
type ServerConfig interface {
// Host returns the domain that the server will run on. Accepts an IPv4 or IPv6 address, in addition to localhost.
Host() string
// Port returns the port that the server will run on. The valid range is [1024, 65535].
Port() int
// User returns the username that connecting clients must use.
User() string
// UserIsSpecified returns true if a user was explicitly provided in the configuration.
UserIsSpecified() bool
// Password returns the password that connecting clients must use.
Password() string
// ReadTimeout returns the read timeout in milliseconds
ReadTimeout() uint64
// WriteTimeout returns the write timeout in milliseconds
WriteTimeout() uint64
// ReadOnly returns whether the server will only accept read statements or all statements.
ReadOnly() bool
// LogLevel returns the level of logging that the server will use.
LogLevel() LogLevel
// LogFormat returns the format of logging that the server will use.
LogFormat() LogFormat
// Autocommit defines the value of the @@autocommit session variable used on every connection
AutoCommit() bool
// DoltTransactionCommit defines the value of the @@dolt_transaction_commit session variable that enables Dolt
// commits to be automatically created when a SQL transaction is committed.
DoltTransactionCommit() bool
// BranchActivityTracking enables or disables the tracking of branch activity for the dolt_branch_activity table
BranchActivityTracking() bool
// DataDir is the path to a directory to use as the data dir, both to create new databases and locate existing ones.
DataDir() string
// CfgDir is the path to a directory to use to store the dolt configuration files.
CfgDir() string
// MaxConnections returns the maximum number of simultaneous connections the server will allow. The default is 1
MaxConnections() uint64
// MaxWaitConnections returns the maximum number of simultaneous connections that the server will allow to block waiting
// for a connection before new connections result in immediate rejection
MaxWaitConnections() uint32
// MaxWaitConnectionsTimeout returns the maximum amount of time that a connection will block waiting for a connection
MaxWaitConnectionsTimeout() time.Duration
// TLSKey returns a path to the servers PEM-encoded private TLS key. "" if there is none.
TLSKey() string
// TLSCert returns a path to the servers PEM-encoded TLS certificate chain. "" if there is none.
TLSCert() string
// CACert returns a path to the servers certificate authority file, or "" if there
// is no CA cert configured.
CACert() string
// RequireSecureTransport is true if the server should reject non-TLS connections.
RequireSecureTransport() bool
// RequireClientCert is true if the server should reject any connections that don't present a certificate. When
// enabled, a client certificate is always required, and if a CA cert is also configured, then the client cert
// will also be verified. Enabling this option also means that non-TLS connections are not allowed.
RequireClientCert() bool
// MaxLoggedQueryLen is the max length of queries written to the logs. Queries longer than this number are truncated.
// If this value is 0 then the query is not truncated and will be written to the logs in its entirety. If the value
// is less than 0 then the queries will be omitted from the logs completely
MaxLoggedQueryLen() int
// ShouldEncodeLoggedQuery determines if logged queries are base64 encoded.
// If true, queries will be logged as base64 encoded strings.
// If false (default behavior), queries will be logged as strings, but newlines and tabs will be replaced with spaces.
ShouldEncodeLoggedQuery() bool
// DisableClientMultiStatements is true if we want the server to not
// process incoming ComQuery packets as if they had multiple queries in
// them, even if the client advertises support for MULTI_STATEMENTS.
DisableClientMultiStatements() bool
// MetricsLabels returns labels that are applied to all prometheus metrics
MetricsLabels() map[string]string
MetricsHost() string
MetricsPort() int
MetricsTLSCert() string
MetricsTLSKey() string
MetricsTLSCA() string
MetricsJwksConfig() *JwksConfig
MetricsJWTRequiredForLocalhost() bool
// PrivilegeFilePath returns the path to the file which contains all needed privilege information in the form of a
// JSON string.
PrivilegeFilePath() string
// BranchControlFilePath returns the path to the file which contains the branch control permissions.
BranchControlFilePath() string
// UserVars is an array containing user specific session variables
UserVars() []UserSessionVars
// SystemVars is a map setting global SQL system variables. For example, `secure_file_priv`.
SystemVars() map[string]interface{}
// JwksConfig is an array containing jwks config
JwksConfig() []JwksConfig
// AllowCleartextPasswords is true if the server should accept cleartext passwords.
AllowCleartextPasswords() bool
// Socket is a path to the unix socket file
Socket() string
// RemotesapiPort is the port to use for serving a remotesapi interface with this sql-server instance.
// A remotesapi interface will allow this sql-server process to be used
// as a dolt remote for things like `clone`, `fetch` and read
// replication.
RemotesapiPort() *int
// RemotesapiReadOnly is true if the remotesapi interface should be read only.
RemotesapiReadOnly() *bool
// MCPPort returns the port for the MCP HTTP server if configured.
MCPPort() *int
// MCPUser returns the SQL user MCP should connect as if configured.
MCPUser() *string
// MCPPassword returns the SQL password MCP should use if configured.
MCPPassword() *string
// MCPDatabase returns the SQL database name MCP should connect to if configured.
MCPDatabase() *string
// ClusterConfig is the configuration for clustering in this sql-server.
ClusterConfig() ClusterConfig
// EventSchedulerStatus is the configuration for enabling or disabling the event scheduler in this server.
EventSchedulerStatus() string
// ValueSet returns whether the value string provided was explicitly set in the config
ValueSet(value string) bool
// AutoGCBehavior defines parameters around how auto-GC works for the running server.
AutoGCBehavior() AutoGCBehavior
// Overrides returns any overrides that are defined. This is primarily used by Doltgres.
Overrides() sql.EngineOverrides
}
// DefaultServerConfig creates a `*ServerConfig` that has all of the options set to their default values.
func DefaultServerConfig() ServerConfig {
return defaultServerConfigYAML()
}
func defaultServerConfigYAML() *YAMLConfig {
return &YAMLConfig{
LogLevelStr: ptr(string(DefaultLogLevel)),
LogFormatStr: ptr(string(DefaultLogFormat)),
MaxQueryLenInLogs: ptr(DefaultMaxLoggedQueryLen),
EncodeLoggedQuery: ptr(DefaultEncodeLoggedQuery),
BehaviorConfig: BehaviorYAMLConfig{
ReadOnly: ptr(DefaultReadOnly),
AutoCommit: ptr(DefaultAutoCommit),
DoltTransactionCommit: ptr(DefaultDoltTransactionCommit),
BranchActivityTracking: ptr(DefaultBranchActivityTracking),
AutoGCBehavior: &AutoGCBehaviorYAMLConfig{
Enable_: ptr(DefaultAutoGCBehaviorEnable),
ArchiveLevel_: ptr(DefaultCompressionLevel),
},
},
UserConfig: UserYAMLConfig{
Name: ptr(""),
Password: ptr(""),
},
ListenerConfig: ListenerYAMLConfig{
HostStr: ptr(DefaultHost),
PortNumber: ptr(DefaultPort),
MaxConnections: ptr(uint64(DefaultMaxConnections)),
BackLog: ptr(uint32(DefaultMaxWaitConnections)),
MaxConnectionsTimeoutMs: ptr(uint64(DefaultMaxWaitConnectionsTimeout.Milliseconds())),
ReadTimeoutMillis: ptr(uint64(DefaultTimeout)),
WriteTimeoutMillis: ptr(uint64(DefaultTimeout)),
AllowCleartextPasswords: ptr(DefaultAllowCleartextPasswords),
},
DataDirStr: ptr(DefaultDataDir),
CfgDirStr: ptr(filepath.Join(DefaultDataDir, DefaultCfgDir)),
PrivilegeFile: ptr(filepath.Join(DefaultDataDir, DefaultCfgDir, DefaultPrivilegeFilePath)),
BranchControlFile: ptr(filepath.Join(DefaultDataDir, DefaultCfgDir, DefaultBranchControlFilePath)),
}
}
// WritableServerConfig is a ServerConfig that support overwriting certain values.
type WritableServerConfig interface {
ServerConfig
// SetUserName sets the username for servers with no other auth established
SetUserName(string)
// SetPassword sets the password for servers with no other auth established
SetPassword(string)
}
type ValidatingServerConfig interface {
ServerConfig
// goldenMysqlConnectionString returns a connection string for a mysql
// instance that can be used to validate query results
GoldenMysqlConnectionString() string
}
// ValidateConfig returns an `error` if any field is not valid.
func ValidateConfig(config ServerConfig) error {
if config.Host() != "localhost" {
ip := net.ParseIP(config.Host())
if ip == nil {
return fmt.Errorf("address is not a valid IP: %v", config.Host())
}
}
if config.Port() < 1024 || config.Port() > 65535 {
return fmt.Errorf("port is not in the range between 1024-65535: %v\n", config.Port())
}
if config.LogLevel().String() == "unknown" {
return fmt.Errorf("loglevel is invalid: %v\n", string(config.LogLevel()))
}
if strings.ToLower(fmt.Sprintf("%v", config.LogFormat())) != string(LogFormat_Text) &&
strings.ToLower(fmt.Sprintf("%v", config.LogFormat())) != string(LogFormat_JSON) {
return fmt.Errorf("logformat is invalid: %v\n", config.LogFormat())
}
if config.RequireSecureTransport() && config.TLSCert() == "" && config.TLSKey() == "" {
return fmt.Errorf("require_secure_transport can only be `true` when a tls_key and tls_cert are provided.")
}
return ValidateClusterConfig(config.ClusterConfig())
}
const (
HostKey = "host"
PortKey = "port"
UserKey = "user"
PasswordKey = "password"
ReadTimeoutKey = "net_read_timeout"
WriteTimeoutKey = "net_write_timeout"
ReadOnlyKey = "read_only"
LogLevelKey = "log_level"
LogFormatKey = "log_format"
AutoCommitKey = "autocommit"
DoltTransactionCommitKey = "dolt_transaction_commit"
BranchActivityTrackingKey = "branch_activity_tracking"
DataDirKey = "data_dir"
CfgDirKey = "cfg_dir"
MaxConnectionsKey = "max_connections"
MaxWaitConnectionsKey = "back_log"
MaxWaitConnectionsTimeoutKey = "max_connections_timeout"
TLSKeyKey = "tls_key"
TLSCertKey = "tls_cert"
RequireSecureTransportKey = "require_secure_transport"
MaxLoggedQueryLenKey = "max_logged_query_len"
ShouldEncodeLoggedQueryKey = "should_encode_logged_query"
DisableClientMultiStatementsKey = "disable_client_multi_statements"
MetricsLabelsKey = "metrics_labels"
MetricsHostKey = "metrics_host"
MetricsPortKey = "metrics_port"
MetricsTLSCertKey = "metrics_tls_cert"
MetricsTLSKeyKey = "metrics_tls_key"
MetricsTLSCAKey = "metrics_tls_ca"
MetricsJwksConfigKey = "metrics_jwks_config"
MetricsJWTRequiredForLocalhostKey = "metrics_jwt_required_for_localhost"
PrivilegeFilePathKey = "privilege_file_path"
BranchControlFilePathKey = "branch_control_file_path"
UserVarsKey = "user_vars"
SystemVarsKey = "system_vars"
JwksConfigKey = "jwks_config"
AllowCleartextPasswordsKey = "allow_cleartext_passwords"
SocketKey = "socket"
RemotesapiPortKey = "remotesapi_port"
RemotesapiReadOnlyKey = "remotesapi_read_only"
ClusterConfigKey = "cluster_config"
EventSchedulerKey = "event_scheduler"
)
type SystemVariableTarget interface {
SetGlobal(ctx *sql.Context, name string, value interface{}) error
}
// ApplySystemVariables sets the global system variables based on the given `ServerConfig`.
func ApplySystemVariables(ctx *sql.Context, cfg ServerConfig, sysVarTarget SystemVariableTarget) error {
if cfg.ValueSet(MaxConnectionsKey) {
err := sysVarTarget.SetGlobal(ctx, "max_connections", cfg.MaxConnections())
if err != nil {
return err
}
}
if cfg.ValueSet(ReadTimeoutKey) {
err := sysVarTarget.SetGlobal(ctx, "net_read_timeout", cfg.ReadTimeout())
if err != nil {
return err
}
}
if cfg.ValueSet(WriteTimeoutKey) {
err := sysVarTarget.SetGlobal(ctx, "net_write_timeout", cfg.WriteTimeout())
if err != nil {
return err
}
}
if cfg.ValueSet(EventSchedulerKey) {
err := sysVarTarget.SetGlobal(ctx, "event_scheduler", cfg.EventSchedulerStatus())
if err != nil {
return err
}
}
return nil
}
func ValidateClusterConfig(config ClusterConfig) error {
if config == nil {
return nil
}
remotes := config.StandbyRemotes()
if len(remotes) == 0 {
return errors.New("cluster config: must supply standby_remotes when supplying cluster configuration.")
}
for i := range remotes {
if remotes[i].Name() == "" {
return fmt.Errorf("cluster: standby_remotes[%d]: name: Cannot be empty", i)
}
if strings.Index(remotes[i].RemoteURLTemplate(), "{database}") == -1 {
return fmt.Errorf("cluster: standby_remotes[%d]: remote_url_template: is \"%s\" but must include the {database} template parameter", i, remotes[i].RemoteURLTemplate())
}
}
if config.BootstrapRole() != "" && config.BootstrapRole() != "primary" && config.BootstrapRole() != "standby" {
return fmt.Errorf("cluster: boostrap_role: is \"%s\" but must be \"primary\" or \"standby\"", config.BootstrapRole())
}
if config.BootstrapEpoch() < 0 {
return fmt.Errorf("cluster: boostrap_epoch: is %d but must be >= 0", config.BootstrapEpoch())
}
if config.RemotesAPIConfig().Port() < 0 || config.RemotesAPIConfig().Port() > 65535 {
return fmt.Errorf("cluster: remotesapi: port: is not in range 0-65535: %d", config.RemotesAPIConfig().Port())
}
if config.RemotesAPIConfig().TLSKey() == "" && config.RemotesAPIConfig().TLSCert() != "" {
return fmt.Errorf("cluster: remotesapi: tls_key: must supply a tls_key if you supply a tls_cert")
}
if config.RemotesAPIConfig().TLSKey() != "" && config.RemotesAPIConfig().TLSCert() == "" {
return fmt.Errorf("cluster: remotesapi: tls_cert: must supply a tls_cert if you supply a tls_key")
}
return nil
}
// ConnectionString returns a Data Source Name (DSN) to be used by go clients for connecting to a running server.
// If unix socket file path is defined in ServerConfig, then `unix` DSN will be returned.
func ConnectionString(config ServerConfig, database string) string {
user := config.User()
if user == "" {
user = "root"
}
var dsn string
if config.Socket() != "" {
dsn = fmt.Sprintf("%v:%v@unix(%v)/%v", user, config.Password(), config.Socket(), database)
} else {
dsn = fmt.Sprintf("%v:%v@tcp(%v:%v)/%v", user, config.Password(), config.Host(), config.Port(), database)
}
if config.AllowCleartextPasswords() {
dsn += "?allowCleartextPasswords=1"
}
return dsn
}
// ConfigInfo returns a summary of some of the config which contains some of the more important information
func ConfigInfo(config ServerConfig) string {
socket := ""
sock, useSock, err := CheckForUnixSocket(config)
if err != nil {
panic(err)
}
if useSock {
socket = fmt.Sprintf(`|S="%v"`, sock)
}
return fmt.Sprintf(`HP="%v:%v"|T="%v"|R="%v"|L="%v"%s`, config.Host(), config.Port(),
config.ReadTimeout(), config.ReadOnly(), config.LogLevel(), socket)
}
func getTLSConfig(cert, key, ca string, requireClientCert bool) (*tls.Config, error) {
if key == "" && cert == "" {
if requireClientCert {
return nil, fmt.Errorf("must supply tls_cert and tls_key when require_client_cert is enabled")
} else {
// No TLS configuration needed
return nil, nil
}
}
c, err := tls.LoadX509KeyPair(cert, key)
if err != nil {
return nil, fmt.Errorf("tls.LoadX509KeyPair(%v, %v) failed: %w", cert, key, err)
}
var caCertPool *x509.CertPool
if ca != "" {
caCertPEM, err := os.ReadFile(ca)
if err != nil {
return nil, fmt.Errorf("unable to read CA file at %s: %w", ca, err)
}
caCertPool = x509.NewCertPool()
if ok := caCertPool.AppendCertsFromPEM(caCertPEM); !ok {
return nil, fmt.Errorf("unable to add CA cert to cert pool")
}
}
clientAuthType := tls.VerifyClientCertIfGiven
if requireClientCert {
// If a CA cert has been specified, then in addition to requiring
// a client cert, also verify it, otherwise allow any client cert.
if ca != "" {
clientAuthType = tls.RequireAndVerifyClientCert
} else {
clientAuthType = tls.RequireAnyClientCert
}
}
return &tls.Config{
Certificates: []tls.Certificate{c},
// tlsVerifyClientCertIfGiven will request a client cert from the client,
// and if provided, will validate it against the specified client CAs.
ClientAuth: clientAuthType,
ClientCAs: caCertPool,
}, nil
}
// LoadTLSConfig loads the certificate chain from config.TLSKey() and config.TLSCert() and returns
// a *tls.Config configured for its use. Returns `nil` if key and cert are `""`.
func LoadTLSConfig(cfg ServerConfig) (*tls.Config, error) {
return getTLSConfig(cfg.TLSCert(), cfg.TLSKey(), cfg.CACert(), cfg.RequireClientCert())
}
func LoadMetricsTLSConfig(cfg ServerConfig) (*tls.Config, error) {
return getTLSConfig(cfg.MetricsTLSCert(), cfg.MetricsTLSKey(), cfg.MetricsTLSCA(), false)
}
// CheckForUnixSocket evaluates ServerConfig for whether the unix socket is to be used or not.
// If user defined socket flag or host is 'localhost', it returns the unix socket file location
// either user-defined or the default if it was not defined.
func CheckForUnixSocket(config ServerConfig) (string, bool, error) {
if config.Socket() != "" {
if runtime.GOOS == "windows" {
return "", false, fmt.Errorf("cannot define unix socket file on Windows")
}
return config.Socket(), true, nil
} else {
// if host is undefined or defined as "localhost" -> unix
if runtime.GOOS != "windows" && config.Host() == "localhost" {
return DefaultUnixSocketFilePath, true, nil
}
}
return "", false, nil
}
type AutoGCBehavior interface {
Enable() bool
ArchiveLevel() int
// IncrementalFileSize > 0 means that chunk files will be periodically written during GC, containing the specified number of chunks.
IncrementalFileSize() uint64
}