Mysql讀寫分離+防止sql注入攻擊「GO原始碼剖析」
一、讀寫分離和防止sql注入的必要性（foreword）
1、 讀寫分離：
一句話定義：讀寫分離，基本的原理是讓主資料庫處理事務性增、改、刪操作（INSERT、UPDATE、DELETE），而從資料庫處理SELECT查詢操作。資料庫複製被用來把事務性操作導致的變更同步到叢集中的從資料庫。
圖1-1 主從讀寫分離示意圖
讀寫分離的好處：
（1） 增加冗餘
（2） 增加了機器的處理能力
（3） 對於讀操作為主的應用，使用讀寫分離是最好的場景，因為可以確保寫的伺服器壓力更小，而讀又可以接受點時間上的延遲。
2、 防止sql注入攻擊：
一句話定義：SQL注入攻擊（SQL Injection），簡稱注入攻擊，是Web開發中最常見的一種安全漏洞。可以用它來從資料庫獲取敏感資訊，或者利用資料庫的特性執行新增使用者，匯出檔案等一系列惡意操作，甚至有可能獲取資料庫乃至系統使用者最高許可權。
二、
最佳解決方案（what）
主庫用讀寫賬戶，從庫用只讀賬戶，建立mysql連結時使用引數interpolateParams=true
package
main
import
（
“database/sql”
“fmt”
_
“github。com/go-sql-driver/mysql”
）
func
main
（）
{
//主庫用讀寫賬戶 建立mysql連結時使用引數interpolateParams=true
db_master
，
err
：=
sql
。
Open
（
“mysql”
，
“master：@tcp（127。0。0。1：3306）/db_test？charset=utf8&interpolateParams=true”
）
if
err
！=
nil
{
panic
（
err
。
Error
（））
//只是舉例，真實使用中需要對錯誤進行處理和返回
}
defer
db_master
。
Close
（）
//從庫用只讀賬戶 建立mysql連結時使用引數interpolateParams=true
db_slave
，
err
：=
sql
。
Open
（
“mysql”
，
“slave：@tcp（127。0。0。1：3306）/db_test？charset=utf8&interpolateParams=true”
）
if
err
！=
nil
{
panic
（
err
。
Error
（））
}
defer
db_slave
。
Close
（）
rows
，
err
：=
db_slave
。
Query
（
“SELECT * FROM tbl_user WHERE user_id = ？”
，
1
）
if
err
！=
nil
{
panic
（
err
。
Error
（））
}
}
三、 幾種方案對比（why）
1、主從都使用
讀寫賬戶
，sql語句採用
字串拼接
方式
（廢棄）
既不能實現主從讀寫分離，也不能防止sql注入攻擊。
2、主從都使用
讀寫賬戶
，sql採用
prepare+execute
的方式
（存在風險）
可以防止sql注入攻擊，但不能實現主從讀寫分離，當併發量上來後主庫壓力會很大，存在風險。
3、主採用
讀寫賬戶
，從採用
只讀賬戶
，sql採用
prepare+execute
的方式
（會報錯）
（1）首先丟擲報錯：
（2）報錯原因分析：
圖1-2 報錯原因分析
錯誤原因：如果prepare的時候sql發給了從庫1，但是execute的時候因為從庫1延時較大，sql命令發給了從庫2，就會報上面這個錯誤（除此之外，預設proxy會隔斷時間切換備機，保證備機都能用到）。
4、主採用
讀寫賬戶
，從採用
只讀賬戶
，建立mysql連結時使用引數
interpolateParams=true
針對方案4，我們需要來看下底層原始碼：
這裡我們需要關注兩部分原始碼：
database/sql和go-sql-driver/mysql
，其中
database/sql
是golang針對資料庫抽象出來的一個標準庫，
go-sql-driver/mysql
是實現database/sql驅動介面的mysql驅動。
（1）我們一般寫查詢語句是這樣的
（舉個栗子）
rows， err ：= db_slave。Query（“SELECT * FROM tbl_user WHERE user_id = ？”， 1）
if err ！= nil {
panic（err。Error（）） //只是舉例，真實使用中需要對錯誤進行處理和返回
}
我們知道prepared statement可以防止sql注入攻擊，上圖這樣的寫法看上去是使用的prepared statement方式，但到底是不是呢？我們繼續往底層走。
（2）database/sql中查詢介面是下面兩個方法（即
Query和QueryRow
）
// Query executes a query that returns rows， typically a SELECT。
// The args are for any placeholder parameters in the query。
func
（
db
*
DB
）
Query
（
query
string
，
args
。。。
interface
{}）
（
*
Rows
，
error
）
{
return
db
。
QueryContext
（
context
。
Background
（），
query
，
args
。。。
）
}
注：Query執行查詢並返回多個數據行，這個查詢通常是一個select，方法中args引數用於填寫查詢語句中包含的佔位符的實際引數。
// QueryRow executes a query that is expected to return at most one row。
// QueryRow always returns a non-nil value。 Errors are deferred until
// Row‘s Scan method is called。
// If the query selects no rows， the *Row’s Scan will return ErrNoRows。
// Otherwise， the *Row‘s Scan scans the first selected row and discards
// the rest。
func
（
db
*
DB
）
QueryRow
（
query
string
，
args
。。。
interface
{}）
*
Row
{
return
db
。
QueryRowContext
（
context
。
Background
（），
query
，
args
。。。
）
}
注：QueryRow與Query方法不同點是，執行一條查詢最多隻會返回一個數據行。
（3）發現寶藏：Query底層執行查詢的策略
從
queryDC
方法來看Query預設是不使用prepared statement方式的，只有在查詢時發生
driver.ErrSkip
錯誤才會啟用prepared statement繼續查詢。
注：從Query到queryDC方法經過的連接獲取和錯誤處理等邏輯不影響我們分析問題，可以忽略。
// queryDC executes a query on the given connection。
// The connection gets released by the releaseConn function。
// The ctx context is from a query method and the txctx context is from an
// optional transaction context。
func
（
db
*
DB
）
queryDC
（
ctx
，
txctx
context
。
Context
，
dc
*
driverConn
，
releaseConn
func
（
error
），
query
string
，
args
［］
interface
{}）
（
*
Rows
，
error
）
{
queryerCtx
，
ok
：=
dc
。
ci
。（
driver
。
QueryerContext
）
var
queryer
driver
。
Queryer
if
！
ok
{
queryer
，
ok
=
dc
。
ci
。（
driver
。
Queryer
）
}
if
ok
{
var
nvdargs
［］
driver
。
NamedValue
var
rowsi
driver
。
Rows
var
err
error
withLock
（
dc
，
func
（）
{
nvdargs
，
err
=
driverArgsConnLocked
（
dc
。
ci
，
nil
，
args
）
if
err
！=
nil
{
return
}
//核心查詢 不使用‘prepared Statement‘來執行
rowsi
，
err
=
ctxDriverQuery
（
ctx
，
queryerCtx
，
queryer
，
query
，
nvdargs
）
}）
if
err
！=
driver
。
ErrSkip
{
//發生錯誤driver。ErrSkip才使用‘prepared Statement‘方式去查詢
if
err
！=
nil
{
//其他錯誤，關閉連結並報錯返回
releaseConn
（
err
）
return
nil
，
err
}
// Note： ownership of dc passes to the *Rows， to be freed
// with releaseConn。
rows
：=
&
Rows
{
dc
：
dc
，
releaseConn
：
releaseConn
，
rowsi
：
rowsi
，
}
rows
。
initContextClose
（
ctx
，
txctx
）
return
rows
，
nil
}
}
//發生錯誤driver。ErrSkip時，才會使用‘Prepared Statement‘方式再次執行查詢
var
si
driver
。
Stmt
var
err
error
withLock
（
dc
，
func
（）
{
si
，
err
=
ctxDriverPrepare
（
ctx
，
dc
。
ci
，
query
）
//prepare
}）
if
err
！=
nil
{
releaseConn
（
err
）
return
nil
，
err
}
ds
：=
&
driverStmt
{
Locker
：
dc
，
si
：
si
}
rowsi
，
err
：=
rowsiFromStatement
（
ctx
，
dc
。
ci
，
ds
，
args
。。。
）
//Statement
if
err
！=
nil
{
ds
。
Close
（）
releaseConn
（
err
）
return
nil
，
err
}
// Note： ownership of ci passes to the *Rows， to be freed
// with releaseConn。
rows
：=
&
Rows
{
dc
：
dc
，
releaseConn
：
releaseConn
，
rowsi
：
rowsi
，
closeStmt
：
ds
，
}
rows
。
initContextClose
（
ctx
，
txctx
）
return
rows
，
nil
}
（4）什麼時候才會報driver。ErrSkip錯誤呢
答案就在go-sql-driver/mysql下的connection。go中真實Query方法中，請看大螢幕：
//mysql query底層實現
func
（
mc
*
mysqlConn
）
query
（
query
string
，
args
［］
driver
。
Value
）
（
*
textRows
，
error
）
{
if
mc
。
closed
。
IsSet
（）
{
errLog
。
（
ErrInvalidConn
）
return
nil
，
driver
。
ErrBadConn
}
if
len
（
args
）
！=
0
{
if
！
mc
。
cfg
。
InterpolateParams
{
//能不能不使用Prepared Statement，就看interpolateParams是不是true了
return
nil
，
driver
。
ErrSkip
}
// try client-side prepare to reduce roundtrip
prepared
，
err
：=
mc
。
interpolateParams
（
query
，
args
）
//在mysql客戶端側使用插值法實現防止sql注入攻擊
if
err
！=
nil
{
return
nil
，
err
}
query
=
prepared
}
// Send command
err
：=
mc
。
writeCommandPacketStr
（
comQuery
，
query
）
if
err
==
nil
{
// Read Result
var
resLen
int
resLen
，
err
=
mc
。
readResultSetHeaderPacket
（）
if
err
==
nil
{
rows
：=
new
（
textRows
）
rows
。
mc
=
mc
if
resLen
==
0
{
rows
。
rs
。
done
=
true
switch
err
：=
rows
。
NextResultSet
（）；
err
{
case
nil
，
io
。
EOF
：
return
rows
，
nil
default
：
return
nil
，
err
}
}
// Columns
rows
。
rs
。
columns
，
err
=
mc
。
readColumns
（
resLen
）
return
rows
，
err
}
}
return
nil
，
mc
。
markBadConn
（
err
）
}
（5） 驅動程式是如何使用插值法防止sql注入攻擊的呢？
我們到interpolateParams方法中一探究竟
（我們只需要關注引數為string的情況）
：
func
（
mc
*
mysqlConn
）
interpolateParams
（
query
string
，
args
［］
driver
。
Value
）
（
string
，
error
）
{
// Number of ？ should be same to len（args）
if
strings
。
Count
（
query
，
“？”
）
！=
len
（
args
）
{
return
“”
，
driver
。
ErrSkip
}
buf
，
err
：=
mc
。
buf
。
takeCompleteBuffer
（）
if
err
！=
nil
{
// can not take the buffer。 Something must be wrong with the connection
errLog
。
（
err
）
return
“”
，
ErrInvalidConn
}
buf
=
buf
［：
0
］
argPos
：=
0
for
i
：=
0
；
i
<
len
（
query
）；
i
++
{
q
：=
strings
。
IndexByte
（
query
［
i
：］，
’？‘
）
if
q
==
-
1
{
buf
=
append
（
buf
，
query
［
i
：］
。。。
）
break
}
buf
=
append
（
buf
，
query
［
i
：
i
+
q
］
。。。
）
i
+=
q
arg
：=
args
［
argPos
］
argPos
++
if
arg
==
nil
{
buf
=
append
（
buf
，
“NULL”
。。。
）
continue
}
switch
v
：=
arg
。（
type
）
{
case
int64
：
buf
=
strconv
。
AppendInt
（
buf
，
v
，
10
）
case
uint64
：
// Handle uint64 explicitly because our custom ConvertValue emits unsigned values
buf
=
strconv
。
AppendUint
（
buf
，
v
，
10
）
case
float64
：
buf
=
strconv
。
AppendFloat
（
buf
，
v
，
’g‘
，
-
1
，
64
）
case
bool
：
if
v
{
buf
=
append
（
buf
，
’1‘
）
}
else
{
buf
=
append
（
buf
，
’0‘
）
}
case
time
。
Time
：
if
v
。
IsZero
（）
{
buf
=
append
（
buf
，
“’0000-00-00‘”
。。。
）
}
else
{
v
：=
v
。
In
（
mc
。
cfg
。
Loc
）
v
=
v
。
Add
（
time
。
Nanosecond
*
500
）
// To round under microsecond
year
：=
v
。
Year
（）
year100
：=
year
/
100
year1
：=
year
%
100
month
：=
v
。
Month
（）
day
：=
v
。
Day
（）
hour
：=
v
。
Hour
（）
minute
：=
v
。
Minute
（）
second
：=
v
。
Second
（）
micro
：=
v
。
Nanosecond
（）
/
1000
buf
=
append
（
buf
，
［］
byte
{
’\‘’
，
digits10
［
year100
］，
digits01
［
year100
］，
digits10
［
year1
］，
digits01
［
year1
］，
‘-’
，
digits10
［
month
］，
digits01
［
month
］，
‘-’
，
digits10
［
day
］，
digits01
［
day
］，
‘ ’
，
digits10
［
hour
］，
digits01
［
hour
］，
‘：’
，
digits10
［
minute
］，
digits01
［
minute
］，
‘：’
，
digits10
［
second
］，
digits01
［
second
］，
}
。。。
）
if
micro
！=
0
{
micro10000
：=
micro
/
10000
micro100
：=
micro
/
100
%
100
micro1
：=
micro
%
100
buf
=
append
（
buf
，
［］
byte
{
‘。’
，
digits10
［
micro10000
］，
digits01
［
micro10000
］，
digits10
［
micro100
］，
digits01
［
micro100
］，
digits10
［
micro1
］，
digits01
［
micro1
］，
}
。。。
）
}
buf
=
append
（
buf
，
‘\’‘
）
}
case
json
。
RawMessage
：
buf
=
append
（
buf
，
’\‘’
）
if
mc
。
status
&
statusNoBackslashEscapes
==
0
{
buf
=
escapeBytesBackslash
（
buf
，
v
）
}
else
{
buf
=
escapeBytesQuotes
（
buf
，
v
）
}
buf
=
append
（
buf
，
‘\’‘
）
case
［］
byte
：
if
v
==
nil
{
buf
=
append
（
buf
，
“NULL”
。。。
）
}
else
{
buf
=
append
（
buf
，
“_binary’”
。。。
）
if
mc
。
status
&
statusNoBackslashEscapes
==
0
{
buf
=
escapeBytesBackslash
（
buf
，
v
）
}
else
{
buf
=
escapeBytesQuotes
（
buf
，
v
）
}
buf
=
append
（
buf
，
‘\’‘
）
}
case
string
：
//我們只需要關注引數為string的情況
buf
=
append
（
buf
，
’\‘’
）
if
mc
。
status
&
statusNoBackslashEscapes
==
0
{
buf
=
escapeStringBackslash
（
buf
，
v
）
}
else
{
buf
=
escapeStringQuotes
（
buf
，
v
）
}
buf
=
append
（
buf
，
‘\’‘
）
default
：
return
“”
，
driver
。
ErrSkip
}
if
len
（
buf
）
+
4
>
mc
。
maxAllowedPacket
{
return
“”
，
driver
。
ErrSkip
}
}
if
argPos
！=
len
（
args
）
{
return
“”
，
driver
。
ErrSkip
}
return
string
（
buf
），
nil
}
原始碼中escapeStringBackslash方法的字面意思是對字串引數轉義，
我們知道轉義特殊字元就是防止sql注入攻擊的有效方法之一
，本著刨根問底的科學探索精神讓我們再進到該方法中一探究竟~
//非常清晰，這裡就是在做字串中特殊字元的轉義編碼
// escapeStringBackslash is similar to escapeBytesBackslash but for string。
func
escapeStringBackslash
（
buf
［］
byte
，
v
string
）
［］
byte
{
pos
：=
len
（
buf
）
buf
=
reserveBuffer
（
buf
，
len
（
v
）
*
2
）
for
i
：=
0
；
i
<
len
（
v
）；
i
++
{
c
：=
v
［
i
］
switch
c
{
case
’\x00‘
：
buf
［
pos
］
=
’\\‘
buf
［
pos
+
1
］
=
’0‘
pos
+=
2
case
’\n‘
：
buf
［
pos
］
=
’\\‘
buf
［
pos
+
1
］
=
’n‘
pos
+=
2
case
’\r‘
：
buf
［
pos
］
=
’\\‘
buf
［
pos
+
1
］
=
’r‘
pos
+=
2
case
’\x1a‘
：
buf
［
pos
］
=
’\\‘
buf
［
pos
+
1
］
=
’Z‘
pos
+=
2
case
’\‘’
：
buf
［
pos
］
=
‘\\’
buf
［
pos
+
1
］
=
‘\’‘
pos
+=
2
case
’“‘
：
buf
［
pos
］
=
’\\‘
buf
［
pos
+
1
］
=
’”‘
pos
+=
2
case
’\\‘
：
buf
［
pos
］
=
’\\‘
buf
［
pos
+
1
］
=
’\\‘
pos
+=
2
default
：
buf
［
pos
］
=
c
pos
++
}
}
return
buf
［：
pos
］
}
這裡就是在做字串中特殊字元的轉義編碼，因此使用引數
interpolateParams=true
可以防止sql注入攻擊，並且因為沒有使用prepared statement方式，當採用主從讀寫分離方式時，也不會再報出方案3中的錯誤，既實現了主從讀寫分離也可以防止sql注入攻擊，是最佳解決方案。
（6）
interpolateParams=true
使用注意事項
可以看到官方文件最後給出一個補充說明，
interpolateParams=true不可以與以下多位元組編碼共同使用（multibyte encodings BIG5, CP932, GB2312, GBK or SJIS），因為他們會引起sql注入脆弱性。
感謝閱讀，如果你感覺本文能幫到你，歡迎點贊、評論和收藏本文，鄙人會再接再厲，為您帶來更多系統架構和機器學習相關的知識分享～