Merge pull request #22 from syncthing/tls

New discovery protocol over HTTPS
This commit is contained in:
Audrius Butkevicius 2015-09-21 08:56:52 +01:00
commit 009b5bc72b
4 changed files with 262 additions and 166 deletions

View File

@ -3,10 +3,10 @@
package main package main
import ( import (
"crypto/tls"
"database/sql" "database/sql"
"flag" "flag"
"log" "log"
"net"
"os" "os"
"time" "time"
@ -21,6 +21,9 @@ var (
statsFile string statsFile string
backend = "ql" backend = "ql"
dsn = getEnvDefault("DISCOSRV_DB_DSN", "memory://discosrv") dsn = getEnvDefault("DISCOSRV_DB_DSN", "memory://discosrv")
certFile = "cert.pem"
keyFile = "key.pem"
debug = false
) )
func main() { func main() {
@ -34,18 +37,23 @@ func main() {
log.SetOutput(os.Stdout) log.SetOutput(os.Stdout)
log.SetFlags(0) log.SetFlags(0)
flag.StringVar(&listen, "listen", ":22027", "Listen address") flag.StringVar(&listen, "listen", ":8443", "Listen address")
flag.IntVar(&lruSize, "limit-cache", lruSize, "Limiter cache entries") flag.IntVar(&lruSize, "limit-cache", lruSize, "Limiter cache entries")
flag.IntVar(&limitAvg, "limit-avg", limitAvg, "Allowed average package rate, per 10 s") flag.IntVar(&limitAvg, "limit-avg", limitAvg, "Allowed average package rate, per 10 s")
flag.IntVar(&limitBurst, "limit-burst", limitBurst, "Allowed burst size, packets") flag.IntVar(&limitBurst, "limit-burst", limitBurst, "Allowed burst size, packets")
flag.StringVar(&statsFile, "stats-file", statsFile, "File to write periodic operation stats to") flag.StringVar(&statsFile, "stats-file", statsFile, "File to write periodic operation stats to")
flag.StringVar(&backend, "db-backend", backend, "Database backend to use") flag.StringVar(&backend, "db-backend", backend, "Database backend to use")
flag.StringVar(&dsn, "db-dsn", dsn, "Database DSN") flag.StringVar(&dsn, "db-dsn", dsn, "Database DSN")
flag.StringVar(&certFile, "cert", certFile, "Certificate file")
flag.StringVar(&keyFile, "key", keyFile, "Key file")
flag.BoolVar(&debug, "debug", debug, "Debug")
flag.Parse() flag.Parse()
addr, _ := net.ResolveUDPAddr("udp", listen) cert, err := tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
log.Fatalln("Failed to load X509 key pair:", err)
}
var err error
db, err := sql.Open(backend, dsn) db, err := sql.Open(backend, dsn)
if err != nil { if err != nil {
log.Fatalln("sql.Open:", err) log.Fatalln("sql.Open:", err)
@ -58,7 +66,8 @@ func main() {
main := suture.NewSimple("main") main := suture.NewSimple("main")
main.Add(&querysrv{ main.Add(&querysrv{
addr: addr, addr: listen,
cert: cert,
db: db, db: db,
prep: prep, prep: prep,
}) })

View File

@ -104,9 +104,9 @@ func postgresCompile(db *sql.DB) (map[string]*sql.Stmt, error) {
"insertAddress": "INSERT INTO Addresses (DeviceID, Seen, Address) VALUES ($1, now(), $2)", "insertAddress": "INSERT INTO Addresses (DeviceID, Seen, Address) VALUES ($1, now(), $2)",
"insertRelay": "INSERT INTO Relays (DeviceID, Seen, Address, Latency) VALUES ($1, now(), $2, $3)", "insertRelay": "INSERT INTO Relays (DeviceID, Seen, Address, Latency) VALUES ($1, now(), $2, $3)",
"insertDevice": "INSERT INTO Devices (DeviceID, Seen) VALUES ($1, now())", "insertDevice": "INSERT INTO Devices (DeviceID, Seen) VALUES ($1, now())",
"selectAddress": "SELECT Address from Addresses WHERE DeviceID=$1 AND Seen > now() - '1 hour'::INTERVAL ORDER BY random() LIMIT 16", "selectAddress": "SELECT Address FROM Addresses WHERE DeviceID=$1 AND Seen > now() - '1 hour'::INTERVAL ORDER BY random() LIMIT 16",
"selectRelay": "SELECT Address, Latency from Relays WHERE DeviceID=$1 AND Seen > now() - '1 hour'::INTERVAL ORDER BY random() LIMIT 16", "selectRelay": "SELECT Address, Latency FROM Relays WHERE DeviceID=$1 AND Seen > now() - '1 hour'::INTERVAL ORDER BY random() LIMIT 16",
"updateRelay": "UPDATE Relays SET Seen=now(), Latency=$3 WHERE DeviceID=$1 AND Address=$2", "updateAddress": "UPDATE Addresses SET Seen=now() WHERE DeviceID=$1 AND Address=$2",
"updateDevice": "UPDATE Devices SET Seen=now() WHERE DeviceID=$1", "updateDevice": "UPDATE Devices SET Seen=now() WHERE DeviceID=$1",
"deleteRelay": "DELETE FROM Relays WHERE DeviceID=$1", "deleteRelay": "DELETE FROM Relays WHERE DeviceID=$1",
} }

View File

@ -80,7 +80,7 @@ func qlCompile(db *sql.DB) (map[string]*sql.Stmt, error) {
"insertDevice": "INSERT INTO Devices (DeviceID, Seen) VALUES ($1, now())", "insertDevice": "INSERT INTO Devices (DeviceID, Seen) VALUES ($1, now())",
"selectAddress": `SELECT Address from Addresses WHERE DeviceID==$1 AND Seen > now() - duration("1h") LIMIT 16`, "selectAddress": `SELECT Address from Addresses WHERE DeviceID==$1 AND Seen > now() - duration("1h") LIMIT 16`,
"selectRelay": `SELECT Address, Latency from Relays WHERE DeviceID==$1 AND Seen > now() - duration("1h") LIMIT 16`, "selectRelay": `SELECT Address, Latency from Relays WHERE DeviceID==$1 AND Seen > now() - duration("1h") LIMIT 16`,
"updateAddress": "UPDATE Addresses Seen=now()WHERE DeviceID==$1 AND Address==$2", "updateAddress": "UPDATE Addresses Seen=now() WHERE DeviceID==$1 AND Address==$2",
"updateDevice": "UPDATE Devices Seen=now() WHERE DeviceID==$1", "updateDevice": "UPDATE Devices Seen=now() WHERE DeviceID==$1",
"deleteRelay": "DELETE FROM Relays WHERE DeviceID==$1", "deleteRelay": "DELETE FROM Relays WHERE DeviceID==$1",
} }

View File

@ -3,213 +3,305 @@
package main package main
import ( import (
"crypto/tls"
"database/sql" "database/sql"
"encoding/binary" "encoding/json"
"fmt"
"io"
"log" "log"
"net" "net"
"net/http"
"net/url" "net/url"
"time" "time"
"github.com/golang/groupcache/lru" "github.com/golang/groupcache/lru"
"github.com/juju/ratelimit" "github.com/juju/ratelimit"
"github.com/syncthing/protocol" "github.com/syncthing/protocol"
"github.com/syncthing/syncthing/lib/discover"
) )
type querysrv struct { type querysrv struct {
addr *net.UDPAddr addr string
db *sql.DB db *sql.DB
prep map[string]*sql.Stmt prep map[string]*sql.Stmt
limiter *lru.Cache limiter *lru.Cache
cert tls.Certificate
listener net.Listener
}
type announcement struct {
Direct []string `json:"direct"`
Relays []annRelay `json:"relays"`
}
type annRelay struct {
URL string `json:"url"`
Latency int `json:"latency"`
} }
func (s *querysrv) Serve() { func (s *querysrv) Serve() {
s.limiter = lru.New(lruSize) s.limiter = lru.New(lruSize)
conn, err := net.ListenUDP("udp", s.addr) tlsCfg := &tls.Config{
Certificates: []tls.Certificate{s.cert},
ClientAuth: tls.RequestClientCert,
SessionTicketsDisabled: true,
MinVersion: tls.VersionTLS12,
CipherSuites: []uint16{
tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,
tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,
tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,
tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,
},
}
http.HandleFunc("/", s.handler)
tlsListener, err := tls.Listen("tcp", s.addr, tlsCfg)
if err != nil { if err != nil {
log.Println("Listen:", err) log.Println("Listen:", err)
return return
} }
// Attempt to set the read and write buffers to 2^24 bytes (16 MB) or as high as s.listener = tlsListener
// possible.
for i := 24; i >= 16; i-- { srv := &http.Server{
if conn.SetReadBuffer(1<<uint(i)) == nil { ReadTimeout: 30 * time.Second,
break WriteTimeout: 10 * time.Second,
} MaxHeaderBytes: 2 << 10,
}
for i := 24; i >= 16; i-- {
if conn.SetWriteBuffer(1<<uint(i)) == nil {
break
}
} }
var buf = make([]byte, 1024) if err := srv.Serve(tlsListener); err != nil {
for { log.Println("Serve:", err)
buf = buf[:cap(buf)]
n, addr, err := conn.ReadFromUDP(buf)
if err != nil {
log.Println("Read:", err)
return
}
if s.limit(addr) {
// Rate limit in effect for source
continue
}
if n < 4 {
log.Printf("Received short packet (%d bytes)", n)
continue
}
buf = buf[:n]
magic := binary.BigEndian.Uint32(buf)
switch magic {
case discover.AnnouncementMagic:
err := s.handleAnnounce(addr, buf)
globalStats.Announce()
if err != nil {
log.Println("Announce:", err)
globalStats.Error()
}
case discover.QueryMagic:
err := s.handleQuery(conn, addr, buf)
globalStats.Query()
if err != nil {
log.Println("Query:", err)
globalStats.Error()
}
default:
globalStats.Error()
}
} }
} }
func (s *querysrv) handler(w http.ResponseWriter, req *http.Request) {
if debug {
log.Println(req.Method, req.URL)
}
remoteAddr, err := net.ResolveTCPAddr("tcp", req.RemoteAddr)
if err != nil {
log.Println("remoteAddr:", err)
http.Error(w, "Internal Server Error", http.StatusInternalServerError)
return
}
if s.limit(remoteAddr.IP) {
if debug {
log.Println(remoteAddr.IP, "is limited")
}
w.Header().Set("Retry-After", "60")
http.Error(w, "Too Many Requests", 429)
return
}
switch req.Method {
case "GET":
s.handleGET(w, req)
case "POST":
s.handlePOST(w, req)
default:
globalStats.Error()
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
}
}
func (s *querysrv) handleGET(w http.ResponseWriter, req *http.Request) {
if req.TLS == nil {
if debug {
log.Println(req.Method, req.URL, "not TLS")
}
globalStats.Error()
http.Error(w, "Forbidden", http.StatusForbidden)
return
}
deviceID, err := protocol.DeviceIDFromString(req.URL.Query().Get("device"))
if err != nil {
if debug {
log.Println(req.Method, req.URL, "bad device param")
}
globalStats.Error()
http.Error(w, "Bad Request", http.StatusBadRequest)
return
}
var ann announcement
ann.Direct, err = s.getAddresses(deviceID)
if err != nil {
log.Println("getAddresses:", err)
globalStats.Error()
http.Error(w, "Internal Server Error", http.StatusInternalServerError)
return
}
ann.Relays, err = s.getRelays(deviceID)
if err != nil {
log.Println("getRelays:", err)
globalStats.Error()
http.Error(w, "Internal Server Error", http.StatusInternalServerError)
return
}
if len(ann.Direct)+len(ann.Relays) == 0 {
globalStats.Error()
http.Error(w, "Not Found", http.StatusNotFound)
return
}
globalStats.Query()
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(ann)
}
func (s *querysrv) handlePOST(w http.ResponseWriter, req *http.Request) {
if req.TLS == nil {
if debug {
log.Println(req.Method, req.URL, "not TLS")
}
globalStats.Error()
http.Error(w, "Forbidden", http.StatusForbidden)
return
}
if len(req.TLS.PeerCertificates) == 0 {
if debug {
log.Println(req.Method, req.URL, "no certificates")
}
globalStats.Error()
http.Error(w, "Forbidden", http.StatusForbidden)
return
}
var ann announcement
if err := json.NewDecoder(req.Body).Decode(&ann); err != nil {
if debug {
log.Println(req.Method, req.URL, err)
}
globalStats.Error()
http.Error(w, "Bad Request", http.StatusBadRequest)
return
}
remoteAddr, err := net.ResolveTCPAddr("tcp", req.RemoteAddr)
if err != nil {
log.Println("remoteAddr:", err)
globalStats.Error()
http.Error(w, "Internal Server Error", http.StatusInternalServerError)
return
}
deviceID := protocol.NewDeviceID(req.TLS.PeerCertificates[0].Raw)
// handleAnnounce returns *two* errors. The first indicates a problem with
// something the client posted to us. We should return a 400 Bad Request
// and not worry about it. The second indicates that the request was fine,
// but something internal fucked up. We should log it and respond with a
// more apologetic 500 Internal Server Error.
userErr, internalErr := s.handleAnnounce(remoteAddr.IP, deviceID, ann.Direct, ann.Relays)
if userErr != nil {
if debug {
log.Println(req.Method, req.URL, userErr)
}
globalStats.Error()
http.Error(w, "Bad Request", http.StatusBadRequest)
return
}
if internalErr != nil {
log.Println("handleAnnounce:", internalErr)
globalStats.Error()
http.Error(w, "Internal Server Error", http.StatusInternalServerError)
return
}
globalStats.Announce()
// TODO: Slowly increase this for stable clients
w.Header().Set("Reannounce-After", "1800")
// We could return the lookup result here, but it's kind of unnecessarily
// expensive to go query the database again so we let the client decide to
// do a lookup if they really care.
w.WriteHeader(http.StatusNoContent)
}
func (s *querysrv) Stop() { func (s *querysrv) Stop() {
panic("stop unimplemented") s.listener.Close()
} }
func (s *querysrv) handleAnnounce(addr *net.UDPAddr, buf []byte) error { func (s *querysrv) handleAnnounce(remote net.IP, deviceID protocol.DeviceID, direct []string, relays []annRelay) (userErr, internalErr error) {
var pkt discover.Announce
err := pkt.UnmarshalXDR(buf)
if err != nil && err != io.EOF {
return err
}
var id protocol.DeviceID
copy(id[:], pkt.This.ID)
if id == protocol.LocalDeviceID {
return fmt.Errorf("Rejecting announce for local device ID from %v", addr)
}
tx, err := s.db.Begin() tx, err := s.db.Begin()
if err != nil { if err != nil {
return err internalErr = err
return
} }
for _, annAddr := range pkt.This.Addresses { defer func() {
// Since we return from a bunch of different places, we handle
// rollback in the defer.
if internalErr != nil || userErr != nil {
tx.Rollback()
}
}()
for _, annAddr := range direct {
uri, err := url.Parse(annAddr) uri, err := url.Parse(annAddr)
if err != nil { if err != nil {
continue userErr = err
return
} }
host, port, err := net.SplitHostPort(uri.Host) host, port, err := net.SplitHostPort(uri.Host)
if err != nil { if err != nil {
continue userErr = err
return
} }
if len(host) == 0 { ip := net.ParseIP(host)
uri.Host = net.JoinHostPort(addr.IP.String(), port) if len(ip) == 0 || ip.IsUnspecified() {
uri.Host = net.JoinHostPort(remote.String(), port)
} }
if err := s.updateAddress(tx, id, uri.String()); err != nil { if err := s.updateAddress(tx, deviceID, uri.String()); err != nil {
tx.Rollback() internalErr = err
return err return
} }
} }
_, err = tx.Stmt(s.prep["deleteRelay"]).Exec(id.String()) _, err = tx.Stmt(s.prep["deleteRelay"]).Exec(deviceID.String())
if err != nil { if err != nil {
tx.Rollback() internalErr = err
return err return
} }
for _, relay := range pkt.This.Relays { for _, relay := range relays {
uri, err := url.Parse(relay.Address) uri, err := url.Parse(relay.URL)
if err != nil { if err != nil {
continue userErr = err
return
} }
_, err = tx.Stmt(s.prep["insertRelay"]).Exec(id.String(), uri, relay.Latency) _, err = tx.Stmt(s.prep["insertRelay"]).Exec(deviceID.String(), uri.String(), relay.Latency)
if err != nil { if err != nil {
tx.Rollback() internalErr = err
return err return
} }
} }
if err := s.updateDevice(tx, id); err != nil { if err := s.updateDevice(tx, deviceID); err != nil {
tx.Rollback() internalErr = err
return err return
} }
return tx.Commit() internalErr = tx.Commit()
return
} }
func (s *querysrv) handleQuery(conn *net.UDPConn, addr *net.UDPAddr, buf []byte) error { func (s *querysrv) limit(remote net.IP) bool {
var pkt discover.Query key := remote.String()
err := pkt.UnmarshalXDR(buf)
if err != nil {
return err
}
var id protocol.DeviceID
copy(id[:], pkt.DeviceID)
addrs, err := s.getAddresses(id)
if err != nil {
return err
}
relays, err := s.getRelays(id)
if err != nil {
return err
}
if len(addrs) > 0 {
ann := discover.Announce{
Magic: discover.AnnouncementMagic,
This: discover.Device{
ID: pkt.DeviceID,
Addresses: addrs,
Relays: relays,
},
}
tb, err := ann.MarshalXDR()
if err != nil {
return fmt.Errorf("QueryV2 response marshal: %v", err)
}
_, err = conn.WriteToUDP(tb, addr)
if err != nil {
return fmt.Errorf("QueryV2 response write: %v", err)
}
globalStats.Answer()
}
return nil
}
func (s *querysrv) limit(addr *net.UDPAddr) bool {
key := addr.IP.String()
bkt, ok := s.limiter.Get(key) bkt, ok := s.limiter.Get(key)
if ok { if ok {
@ -279,26 +371,21 @@ func (s *querysrv) getAddresses(device protocol.DeviceID) ([]string, error) {
return res, nil return res, nil
} }
func (s *querysrv) getRelays(device protocol.DeviceID) ([]discover.Relay, error) { func (s *querysrv) getRelays(device protocol.DeviceID) ([]annRelay, error) {
rows, err := s.prep["selectRelay"].Query(device.String()) rows, err := s.prep["selectRelay"].Query(device.String())
if err != nil { if err != nil {
return nil, err return nil, err
} }
var res []discover.Relay var res []annRelay
for rows.Next() { for rows.Next() {
var addr string var rel annRelay
var latency int32
err := rows.Scan(&addr, &latency) err := rows.Scan(&rel.URL, &rel.Latency)
if err != nil { if err != nil {
log.Println("Scan:", err) return nil, err
continue
} }
res = append(res, discover.Relay{ res = append(res, rel)
Address: addr,
Latency: latency,
})
} }
return res, nil return res, nil