Fix query construction

`sqlutil.INList` repeats the first value of the list and leaves the last value off the resulting string. Fixed this bug and replaced uses of unescaped `IN` list creation with calls to `sqlutil.INList` where appropriate.

Adjusted metrics and heartbeat to use driver-level parameter interpolation instead of custom escaping, deprecating several function in `sqlutil`. Removed the use of `sqlutil.INList` in existing blip code but marked as deprecated.

Author: Areson

dbconn/factory.go

@@ -116,7 +116,7 @@ func (f factory) Make(cfg blip.ConfigMonitor) (*sql.DB, string, error) {
 	// ----------------------------------------------------------------------
 	// Load TLS
 
-	params := []string{"parseTime=true"}
+	params := []string{"parseTime=true", "interpolateParams=true"}
 
 	// Go says "either ServerName or InsecureSkipVerify must be specified".
 	// This is a pathological case: socket and TLS but no hostname to verify

dbconn/factory_test.go

@@ -54,7 +54,7 @@ func TestConnect(t *testing.T) {
 	defer db.Close()
 
 	// Make returns a print-safe DSN: password ("test") replaced with "..."
-	expectDSN := fmt.Sprintf("%s:...@tcp(%s)/?parseTime=true", cfg.Username, cfg.Hostname)
+	expectDSN := fmt.Sprintf("%s:...@tcp(%s)/?interpolateParams=true&parseTime=true", cfg.Username, cfg.Hostname)
 	if dsn != expectDSN {
 		t.Errorf("got DSN '%s', expected '%s'", dsn, expectDSN)
 	}

heartbeat/reader.go

@@ -56,6 +56,7 @@ type BlipReader struct {
 	isRepl   bool
 	event    event.MonitorReceiver
 	query    string
+	params   []interface{}
 }
 
 type BlipReaderArgs struct {
@@ -91,13 +92,16 @@ func NewBlipReader(args BlipReaderArgs) *BlipReader {
 	var where string
 	if r.srcId != "" {
 		blip.Debug("%s: heartbeat from source %s", r.monitorId, r.srcId)
-		where = "WHERE src_id='" + r.srcId + "'" // default
+		where = "WHERE src_id=?" // default
+		r.params = []interface{}{r.srcId}
 	} else if r.srcRole != "" {
 		blip.Debug("%s: heartbeat from role %s", r.monitorId, r.srcRole)
-		where = "WHERE src_role='" + r.srcRole + "' ORDER BY ts DESC LIMIT 1"
+		where = "WHERE src_role=? ORDER BY ts DESC LIMIT 1"
+		r.params = []interface{}{r.srcRole}
 	} else {
 		blip.Debug("%s: heartbeat from latest (max ts)", r.monitorId)
-		where = "WHERE src_id != '" + args.MonitorId + "' ORDER BY ts DESC LIMIT 1"
+		where = "WHERE src_id != ? ORDER BY ts DESC LIMIT 1"
+		r.params = []interface{}{args.MonitorId}
 	}
 	if r.replCheck != "" {
 		cols[4] = "@@" + r.replCheck
@@ -136,7 +140,7 @@ func (r *BlipReader) run() {
 		}
 
 		ctx, cancel = context.WithTimeout(context.Background(), ReadTimeout)
-		err = r.db.QueryRowContext(ctx, r.query).Scan(&now, &last, &freq, &srcId, &isRepl)
+		err = r.db.QueryRowContext(ctx, r.query, r.params...).Scan(&now, &last, &freq, &srcId, &isRepl)
 		cancel()
 		if err != nil {
 			blip.Debug("%s: %v", r.monitorId, err)

heartbeat/writer.go

@@ -84,18 +84,19 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error {
 	// This must be done else the simpler UPDATE statements below, which is the
 	// real heartbeat, will fail because there's no match row.
 	var ping string
+	var params []interface{}
 	if w.srcRole != "" {
-		ping = fmt.Sprintf("INSERT INTO %s (src_id, src_role, ts, freq) VALUES ('%s', '%s', NOW(3), %d) ON DUPLICATE KEY UPDATE ts=NOW(3), freq=%d, src_role='%s'",
-			w.table, w.srcId, w.srcRole, w.freq.Milliseconds(), w.freq.Milliseconds(), w.srcRole)
+		ping = fmt.Sprintf("INSERT INTO %s (src_id, src_role, ts, freq) VALUES (?, ?, NOW(3), ?) ON DUPLICATE KEY UPDATE ts=NOW(3), freq=?, src_role=?", w.table)
+		params = []interface{}{w.srcId, w.srcRole, w.freq.Milliseconds(), w.freq.Milliseconds(), w.srcRole}
 	} else {
-		ping = fmt.Sprintf("INSERT INTO %s (src_id, src_role, ts, freq) VALUES ('%s', NULL, NOW(3), %d) ON DUPLICATE KEY UPDATE ts=NOW(3), freq=%d, src_role=NULL",
-			w.table, w.monitorId, w.freq.Milliseconds(), w.freq.Milliseconds())
+		ping = fmt.Sprintf("INSERT INTO %s (src_id, src_role, ts, freq) VALUES (?, NULL, NOW(3), ?) ON DUPLICATE KEY UPDATE ts=NOW(3), freq=?, src_role=NULL", w.table)
+		params = []interface{}{w.monitorId, w.freq.Milliseconds(), w.freq.Milliseconds()}
 	}
 	blip.Debug("%s: first heartbeat: %s", w.monitorId, ping)
 	for {
 		status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "first insert")
 		ctx, cancel = context.WithTimeout(context.Background(), WriteTimeout)
-		_, err = w.db.ExecContext(ctx, ping)
+		_, err = w.db.ExecContext(ctx, ping, params...)
 		cancel()
 		if err == nil { // success
 			status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "sleep")
@@ -127,14 +128,14 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error {
 	// to void 2 wasted round trips: prep (waste), exec, close (waste).
 	// This risk of SQL injection is miniscule because both table and monitorId
 	// are sanitized, and Blip should only have write privs on its heartbeat table.
-	ping = fmt.Sprintf("UPDATE %s SET ts=NOW(3) WHERE src_id='%s'", w.table, w.srcId)
+	ping = fmt.Sprintf("UPDATE %s SET ts=NOW(3) WHERE src_id=?", w.table)
 	blip.Debug("%s: heartbeat: %s", w.monitorId, ping)
 	for {
 		time.Sleep(w.freq)
 
 		status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "write")
 		ctx, cancel = context.WithTimeout(context.Background(), WriteTimeout)
-		_, err = w.db.ExecContext(ctx, ping)
+		_, err = w.db.ExecContext(ctx, ping, w.srcId)
 		cancel()
 		if err != nil {
 			blip.Debug("%s: %s", w.monitorId, err.Error())

metrics/innodb/innodb.go

@@ -42,16 +42,18 @@ const (
 // InnoDB collects metrics for the innodb domain. The source is
 // information_schema.innodb_metrics.
 type InnoDB struct {
-	db    *sql.DB
-	query map[string]string
+	db     *sql.DB
+	query  map[string]string
+	params map[string][]interface{}
 }
 
 var _ blip.Collector = &InnoDB{}
 
 func NewInnoDB(db *sql.DB) *InnoDB {
 	return &InnoDB{
-		db:    db,
-		query: map[string]string{},
+		db:     db,
+		query:  map[string]string{},
+		params: map[string][]interface{}{},
 	}
 }
 
@@ -95,18 +97,21 @@ LEVEL:
 		switch all {
 		case "all":
 			c.query[level.Name] = baseQuery
+			c.params[level.Name] = []interface{}{}
 		case "enabled":
 			c.query[level.Name] = baseQuery + " WHERE status='enabled'"
+			c.params[level.Name] = []interface{}{}
 		default:
-			c.query[level.Name] = baseQuery + " WHERE name IN (" + sqlutil.INList(dom.Metrics, "'") + ")"
+			c.query[level.Name] = baseQuery + " WHERE name IN (" + sqlutil.PlaceholderList(len(dom.Metrics)) + ")"
+			c.params[level.Name] = sqlutil.ToInterfaceArray(dom.Metrics)
 		}
 		blip.Debug("%s: innodb metrics at %s: %s", plan.MonitorId, level.Name, c.query[level.Name])
 	}
 	return nil, nil
 }
 
 func (c *InnoDB) Collect(ctx context.Context, levelName string) ([]blip.MetricValue, error) {
-	rows, err := c.db.QueryContext(ctx, c.query[levelName])
+	rows, err := c.db.QueryContext(ctx, c.query[levelName], c.params[levelName]...)
 	if err != nil {
 		return nil, err
 	}

metrics/repl.lag/lag.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/cashapp/blip"
 	"github.com/cashapp/blip/heartbeat"
-	"github.com/cashapp/blip/sqlutil"
 )
 
 const (
@@ -207,12 +206,28 @@ LEVEL:
 
 		c.dropNotAReplica[levelName] = !blip.Bool(dom.Options[OPT_REPORT_NOT_A_REPLICA])
 		c.defaultChannelNameOverrides[levelName] = dom.Options[OPT_DEFAULT_CHANNEL_NAME]
-		c.replCheck = sqlutil.CleanObjectName(dom.Options[OPT_REPL_CHECK]) // @todo sanitize better
+		if err := c.verifyReplCheck(ctx, dom.Options[OPT_REPL_CHECK]); err != nil {
+			return cleanup, err
+		}
+		c.replCheck = dom.Options[OPT_REPL_CHECK]
 	}
 
 	return cleanup, nil
 }
 
+func (c *Lag) verifyReplCheck(ctx context.Context, variable string) error {
+	if variable == "" {
+		return nil
+	}
+
+	row := c.db.QueryRowContext(ctx, "SHOW GLOBAL VARIABLES LIKE ?", variable)
+	var value string
+	if err := row.Scan(&value, &value); err != nil {
+		return fmt.Errorf("failed to verify replication check variable %s: %s", variable, err)
+	}
+	return nil
+}
+
 func (c *Lag) Collect(ctx context.Context, levelName string) ([]blip.MetricValue, error) {
 	switch c.lagWriterIn[levelName] {
 	case LAG_WRITER_BLIP:

metrics/size.database/database.go

@@ -24,17 +24,19 @@ const (
 type Database struct {
 	db *sql.DB
 	// --
-	query map[string]string // keyed on level
-	total map[string]bool   // keyed on level
+	query  map[string]string // keyed on level
+	params map[string][]interface{}
+	total  map[string]bool // keyed on level
 }
 
 var _ blip.Collector = &Database{}
 
 func NewDatabase(db *sql.DB) *Database {
 	return &Database{
-		db:    db,
-		query: map[string]string{},
-		total: map[string]bool{},
+		db:     db,
+		query:  map[string]string{},
+		params: map[string][]interface{}{},
+		total:  map[string]bool{},
 	}
 }
 
@@ -97,11 +99,12 @@ LEVEL:
 		if !ok {
 			continue LEVEL // not collected in this level
 		}
-		q, err := DataSizeQuery(dom.Options, c.Help())
+		q, p, err := DataSizeQuery(dom.Options, c.Help())
 		if err != nil {
 			return nil, err
 		}
 		c.query[level.Name] = q
+		c.params[level.Name] = p
 
 		if dom.Options[OPT_TOTAL] == "yes" {
 			c.total[level.Name] = true
@@ -116,7 +119,7 @@ func (c *Database) Collect(ctx context.Context, levelName string) ([]blip.Metric
 		return nil, nil // not collected in this level
 	}
 
-	rows, err := c.db.QueryContext(ctx, q)
+	rows, err := c.db.QueryContext(ctx, q, c.params[levelName]...)
 	if err != nil {
 		return nil, err
 	}

metrics/size.database/query.go

@@ -10,7 +10,7 @@ import (
 	"github.com/cashapp/blip/sqlutil"
 )
 
-func DataSizeQuery(set map[string]string, def blip.CollectorHelp) (string, error) {
+func DataSizeQuery(set map[string]string, def blip.CollectorHelp) (string, []interface{}, error) {
 	cols := ""
 	groupBy := ""
 	if val := set[OPT_TOTAL]; val == "only" {
@@ -26,32 +26,42 @@ func DataSizeQuery(set map[string]string, def blip.CollectorHelp) (string, error
 		like = true
 	}
 
+	var params []interface{}
+
 	where := ""
 	if include := set[OPT_INCLUDE]; include != "" {
-		o := sqlutil.ObjectList(include, "'")
+		o := strings.Split(include, ",")
+		params = make([]interface{}, 0, len(o))
+
 		if like {
 			for i := range o {
-				o[i] = "table_schema LIKE " + o[i]
+				params = append(params, o[i])
+				o[i] = "table_schema LIKE ?"
 			}
 			where = strings.Join(o, " OR ")
 		} else {
-			where = fmt.Sprintf("table_schema IN (%s)", strings.Join(o, ","))
+			where = fmt.Sprintf("table_schema IN (%s)", sqlutil.PlaceholderList(len(o)))
+			params = sqlutil.ToInterfaceArray(o)
 		}
 	} else {
 		exclude := set[OPT_EXCLUDE]
 		if exclude == "" {
 			exclude = def.Options[OPT_EXCLUDE].Default
 		}
-		o := sqlutil.ObjectList(exclude, "'")
+		o := strings.Split(exclude, ",")
+		params = make([]interface{}, 0, len(o))
+
 		if like {
 			for i := range o {
-				o[i] = "table_schema NOT LIKE " + o[i]
+				params = append(params, o[i])
+				o[i] = "table_schema NOT LIKE ?"
 			}
 			where = strings.Join(o, " AND ")
 		} else {
-			where = fmt.Sprintf("table_schema NOT IN (%s)", strings.Join(o, ","))
+			where = fmt.Sprintf("table_schema NOT IN (%s)", sqlutil.PlaceholderList(len(o)))
+			params = sqlutil.ToInterfaceArray(o)
 		}
 	}
 
-	return "SELECT " + cols + " FROM information_schema.tables WHERE " + where + groupBy, nil
+	return "SELECT " + cols + " FROM information_schema.tables WHERE " + where + groupBy, params, nil
 }