Martchus
/
syncthing
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

New discovery protocol over HTTPS

This commit is contained in:
Jakob Borg 2015-09-13 11:44:33 +02:00
parent cbe24d0c61
commit 9b541a28e6
4 changed files with 262 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
cmd/discosrv/main.go
View File

@ -3,10 +3,10 @@
package main
import (
"crypto/tls"
"database/sql"
"flag"
"log"
"net"
"os"
"time"
@ -21,6 +21,9 @@ var (
statsFile string
backend = "ql"
dsn = getEnvDefault("DISCOSRV_DB_DSN", "memory://discosrv")
certFile = "cert.pem"
keyFile = "key.pem"
debug = false
)
func main() {
@ -34,18 +37,23 @@ func main() {
log.SetOutput(os.Stdout)
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(&limitAvg, "limit-avg", limitAvg, "Allowed average package rate, per 10 s")
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(&backend, "db-backend", backend, "Database backend to use")
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()
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)
if err != nil {
log.Fatalln("sql.Open:", err)
@ -58,7 +66,8 @@ func main() {
main := suture.NewSimple("main")
main.Add(&querysrv{
addr: addr,
addr: listen,
cert: cert,
db: db,
prep: prep,
})

6
cmd/discosrv/psql.go
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)",
"insertRelay": "INSERT INTO Relays (DeviceID, Seen, Address, Latency) VALUES ($1, now(), $2, $3)",
"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",
"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",
"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",
"updateAddress": "UPDATE Addresses SET Seen=now() WHERE DeviceID=$1 AND Address=$2",
"updateDevice": "UPDATE Devices SET Seen=now() WHERE DeviceID=$1",
"deleteRelay": "DELETE FROM Relays WHERE DeviceID=$1",
}

2
cmd/discosrv/ql.go
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())",
"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`,
"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",
"deleteRelay": "DELETE FROM Relays WHERE DeviceID==$1",
}

401
cmd/discosrv/querysrv.go
View File

@ -3,213 +3,305 @@
package main
import (
"crypto/tls"
"database/sql"
"encoding/binary"
"fmt"
"io"
"encoding/json"
"log"
"net"
"net/http"
"net/url"
"time"
"github.com/golang/groupcache/lru"
"github.com/juju/ratelimit"
"github.com/syncthing/protocol"
"github.com/syncthing/syncthing/lib/discover"
)
type querysrv struct {
addr *net.UDPAddr
db *sql.DB
prep map[string]*sql.Stmt
limiter *lru.Cache
addr string
db *sql.DB
prep map[string]*sql.Stmt
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() {
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 {
log.Println("Listen:", err)
return
}
// Attempt to set the read and write buffers to 2^24 bytes (16 MB) or as high as
// possible.
for i := 24; i >= 16; i-- {
if conn.SetReadBuffer(1<<uint(i)) == nil {
break
}
}
for i := 24; i >= 16; i-- {
if conn.SetWriteBuffer(1<<uint(i)) == nil {
break
}
s.listener = tlsListener
srv := &http.Server{
ReadTimeout: 30 * time.Second,
WriteTimeout: 10 * time.Second,
MaxHeaderBytes: 2 << 10,
}
var buf = make([]byte, 1024)
for {
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()
}
if err := srv.Serve(tlsListener); err != nil {
log.Println("Serve:", err)
}
}
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() {
panic("stop unimplemented")
s.listener.Close()
}
func (s *querysrv) handleAnnounce(addr *net.UDPAddr, buf []byte) 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)
}
func (s *querysrv) handleAnnounce(remote net.IP, deviceID protocol.DeviceID, direct []string, relays []annRelay) (userErr, internalErr error) {
tx, err := s.db.Begin()
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)
if err != nil {
continue
userErr = err
return
}
host, port, err := net.SplitHostPort(uri.Host)
if err != nil {
continue
userErr = err
return
}
if len(host) == 0 {
uri.Host = net.JoinHostPort(addr.IP.String(), port)
ip := net.ParseIP(host)
if len(ip) == 0 || ip.IsUnspecified() {
uri.Host = net.JoinHostPort(remote.String(), port)
}
if err := s.updateAddress(tx, id, uri.String()); err != nil {
tx.Rollback()
return err
if err := s.updateAddress(tx, deviceID, uri.String()); err != nil {
internalErr = err
return
}
}
_, err = tx.Stmt(s.prep["deleteRelay"]).Exec(id.String())
_, err = tx.Stmt(s.prep["deleteRelay"]).Exec(deviceID.String())
if err != nil {
tx.Rollback()
return err
internalErr = err
return
}
for _, relay := range pkt.This.Relays {
uri, err := url.Parse(relay.Address)
for _, relay := range relays {
uri, err := url.Parse(relay.URL)
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 {
tx.Rollback()
return err
internalErr = err
return
}
}
if err := s.updateDevice(tx, id); err != nil {
tx.Rollback()
return err
if err := s.updateDevice(tx, deviceID); err != nil {
internalErr = err
return
}
return tx.Commit()
internalErr = tx.Commit()
return
}
func (s *querysrv) handleQuery(conn *net.UDPConn, addr *net.UDPAddr, buf []byte) error {
var pkt discover.Query
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()
func (s *querysrv) limit(remote net.IP) bool {
key := remote.String()
bkt, ok := s.limiter.Get(key)
if ok {
@ -279,26 +371,21 @@ func (s *querysrv) getAddresses(device protocol.DeviceID) ([]string, error) {
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())
if err != nil {
return nil, err
}
var res []discover.Relay
var res []annRelay
for rows.Next() {
var addr string
var latency int32
var rel annRelay
err := rows.Scan(&addr, &latency)
err := rows.Scan(&rel.URL, &rel.Latency)
if err != nil {
log.Println("Scan:", err)
continue
return nil, err
}
res = append(res, discover.Relay{
Address: addr,
Latency: latency,
})
res = append(res, rel)
}
return res, nil