X go/neo: Add SSL support

go/neo/client.go

@@ -615,12 +615,56 @@ func openClientByURL(ctx context.Context, u *url.URL, opt *zodb.DriverOptions) (
 		return nil, zodb.InvalidTid, fmt.Errorf("cluster name not specified")
 	}
 
+	qv, err := url.ParseQuery(u.RawQuery)
+	if err != nil {
+		return nil, zodb.InvalidTid, err
+	}
+	q := map[string]string{}
+	for k, vv := range qv {
+		if len(vv) == 0 {
+			return nil, zodb.InvalidTid, fmt.Errorf("parameter %q without value", k)
+		}
+		if len(vv) != 1 {
+			return nil, zodb.InvalidTid, fmt.Errorf("duplicate parameter %q ", k)
+		}
+		q[k] = vv[0]
+	}
+
+	qpop := func(k string) string {
+		v := q[k]
+		delete(q, k)
+		return v
+	}
+
+	ssl  := false
+	ca   := qpop("ca")
+	cert := qpop("cert")
+	key  := qpop("key")
+
+	if len(q) != 0 {
+		return nil, zodb.InvalidTid, fmt.Errorf("invalid query: %v", q)
+	}
+
+	if ca != "" || cert != "" || key != "" {
+		if !(ca != "" && cert != "" && key != "") {
+			return nil, zodb.InvalidTid, fmt.Errorf("incomplete ca/cert/key provided")
+		}
+		ssl = true
+	}
+
+
 	if !opt.ReadOnly {
 		return nil, zodb.InvalidTid, fmt.Errorf("TODO write mode not implemented")
 	}
 
-	// XXX check/use other url fields
-	net := xnet.NetPlain("tcp")	// TODO + TLS; not only "tcp" ?
+	net := xnet.NetPlain("tcp")	// TODO not only "tcp" ?
+	if ssl {
+		tlsCfg, err := tlsForSSL(ca, cert, key)
+		if err != nil {
+			return nil, zodb.InvalidTid, err
+		}
+		net = xnet.NetTLS(net, tlsCfg)
+	}
 
 	c := NewClient(u.User.Username(), u.Host, net)
 	c.watchq = opt.Watchq

go/neo/client_test.go

@@ -26,6 +26,7 @@ import (
 	"net/url"
 	"os"
 	"os/exec"
+	"strings"
 	"testing"
 	"time"
 
@@ -38,8 +39,10 @@ import (
 
 // NEOSrv represents running NEO server.
 type NEOSrv interface {
-	ClusterName() string // name of the cluster
-	MasterAddr()  string // address of the master
+	// XXX kill or restore?
+	//ClusterName() string // name of the cluster
+	//MasterAddr()  string // address of the master
+	ZUrl() string // zurl to access this NEO server
 
 	Bugs() []string // list of known server bugs
 }
@@ -57,6 +60,11 @@ type NEOPySrv struct {
 	errExit     error         // error from Wait
 
 	masterAddr string // address of master in spawned cluster
+
+	// CA/Cert/Key used for SSL exchange. Empty if SSL is not used
+	CA   string
+	Cert string
+	Key  string
 }
 
 func (_ *NEOPySrv) Bugs() []string {
@@ -71,6 +79,8 @@ type NEOPyOptions struct {
 	// nreplica
 
 	// name
+
+	SSL bool // whether to use SSL for node-node exchange
 }
 
 // StartNEOPySrv starts NEO/py server for clusterName NEO database located in workdir/.
@@ -90,8 +100,20 @@ func StartNEOPySrv(workdir, clusterName string, opt NEOPyOptions) (_ *NEOPySrv,
 	}
 
 	n := &NEOPySrv{workdir: workdir, clusterName: clusterName, cancel: cancel, done: make(chan struct{})}
+	if opt.SSL {
+		npytests := "../../neo/tests/"
+		n.CA   = npytests + "ca.crt"
+		n.Cert = npytests + "node.crt"
+		n.Key  = npytests + "node.key"
+	}
+
 	// XXX $PYTHONPATH to top, so that `import neo` works?
-	n.pysrv = xexec.Command("./py/runneo.py", workdir, clusterName) // XXX +opt
+	n.pysrv = xexec.Command("./py/runneo.py", workdir, clusterName)
+	if opt.SSL {
+		n.pysrv.Args = append(n.pysrv.Args, "ca="  +n.CA)
+		n.pysrv.Args = append(n.pysrv.Args, "cert="+n.Cert)
+		n.pysrv.Args = append(n.pysrv.Args, "key=" +n.Key)
+	}
 	n.opt = opt
 	// $TEMP -> workdir  (else NEO/py creates another one for e.g. coverage)
 	n.pysrv.Env = append(os.Environ(), "TEMP="+workdir)
@@ -152,6 +174,23 @@ func (n *NEOPySrv) MasterAddr() string {
 	return n.masterAddr
 }
 
+func (n *NEOPySrv) ZUrl() string {
+	zurl := fmt.Sprintf("neo://%s@%s", n.ClusterName(), n.MasterAddr())
+	argv := []string{}
+	if n.opt.SSL {
+		argv = append(argv, "ca="  +url.QueryEscape(n.CA))
+		argv = append(argv, "cert="+url.QueryEscape(n.Cert))
+		argv = append(argv, "key=" +url.QueryEscape(n.Key))
+	}
+
+	if len(argv) != 0 {
+		zurl += "?"
+		zurl += strings.Join(argv, "&")
+	}
+
+	return zurl
+}
+
 func (n *NEOPySrv) Close() (err error) {
 	defer xerr.Contextf(&err, "stopneo %s", n.workdir)
 
@@ -194,16 +233,22 @@ func withNEOSrv(t *testing.T, f func(t *testing.T, nsrv NEOSrv), optv ...tOption
 		f(work)
 	}
 
-	// TODO + all variants with nreplic=X, npartition=Y, nmaster=Z, ... ?
+	// TODO + all variants with nreplica=X, npartition=Y, nmaster=Z, ... ?
+
+	for _, ssl := range []bool{false, true} {
+		kind := ""
+		if ssl { kind = "ssl" } else { kind = "!ssl" }
 
 		// NEO/py
-	t.Run("py", func(t *testing.T) {
+		t.Run("py/"+kind, func(t *testing.T) {
 			t.Helper()
 			// XXX needpy
 			inWorkDir(t, func(workdir string) {
 				X := xtesting.FatalIf(t)
 
-			npy, err := StartNEOPySrv(workdir, "1", NEOPyOptions{}); X(err)
+				npy, err := StartNEOPySrv(workdir, "1", NEOPyOptions{
+					SSL: ssl,
+				}); X(err)
 				defer func() {
 					err := npy.Close(); X(err)
 				}()
@@ -213,8 +258,16 @@ func withNEOSrv(t *testing.T, f func(t *testing.T, nsrv NEOSrv), optv ...tOption
 						"from neo.scripts.neomigrate import main; main()",
 						"-q",
 						"-c", npy.ClusterName(),
+					)
+					if ssl {
+						cmd.Args = append(cmd.Args, "--ca",   npy.CA)
+						cmd.Args = append(cmd.Args, "--cert", npy.Cert)
+						cmd.Args = append(cmd.Args, "--key",  npy.Key)
+					}
+					cmd.Args = append(cmd.Args,
 						opt.Preload,
-					npy.MasterAddr())
+						npy.MasterAddr(),
+					)
 					cmd.Stdin  = nil
 					cmd.Stdout = os.Stdout
 					cmd.Stderr = os.Stderr
@@ -227,6 +280,7 @@ func withNEOSrv(t *testing.T, f func(t *testing.T, nsrv NEOSrv), optv ...tOption
 
 
 		// TODO NEO/go
+	}
 }
 
 // withNEO tests f on all kinds of NEO servers connected to by NEO client.
@@ -235,7 +289,7 @@ func withNEO(t *testing.T, f func(t *testing.T, nsrv NEOSrv, ndrv *Client), optv
 	withNEOSrv(t, func(t *testing.T, nsrv NEOSrv) {
 		t.Helper()
 		X := xtesting.FatalIf(t)
-		ndrv, _, err := neoOpen(fmt.Sprintf("neo://%s@%s", nsrv.ClusterName(), nsrv.MasterAddr()),
+		ndrv, _, err := neoOpen(nsrv.ZUrl(),
 					&zodb.DriverOptions{ReadOnly: true}); X(err)
 		defer func() {
 			err := ndrv.Close(); X(err)
@@ -271,7 +325,7 @@ func TestLoad(t *testing.T) {
 
 func TestWatch(t *testing.T) {
 	withNEOSrv(t, func(t *testing.T, nsrv NEOSrv) {
-		xtesting.DrvTestWatch(t, fmt.Sprintf("neo://%s@%s", nsrv.ClusterName(), nsrv.MasterAddr()), openClientByURL)
+		xtesting.DrvTestWatch(t, nsrv.ZUrl(), openClientByURL)
 	})
 }
 

go/neo/py/runneo.py

@@ -20,7 +20,7 @@
 # See https://www.nexedi.com/licensing for rationale and options.
 """runneo.py runs NEO/py cluster for NEO/go testing.
 
-Usage: runneo.py <workdir> <cluster-name>   XXX + (**kw for NEOCluster)
+Usage: runneo.py <workdir> <cluster-name>   [k1=v1] [k2=v2] ...
 
 <workdir>/ready is created with address of master after spawned cluster becomes
 operational.
@@ -40,7 +40,14 @@ def main():
     clusterName = sys.argv[2]
     readyf      = workdir + "/ready"
 
-    def sinfo(msg): return "I: runneo.py: %s/%s: %s" % (workdir, clusterName, msg)
+    kw = {}
+    for arg in sys.argv[3:]:
+        k, v = arg.split('=')
+        kw[k] = v
+
+
+    flags = ''
+    def sinfo(msg): return "I: runneo.py: %s/%s%s: %s" % (workdir, clusterName, flags, msg)
     def info(msg):  print(sinfo(msg))
 
     # SIGTERM -> exit gracefully, so that defers are run
@@ -48,7 +55,21 @@ def main():
         raise SystemExit(sinfo("terminated"))
     signal(SIGTERM, _)
 
-    cluster = NEOCluster([clusterName], adapter='SQLite', name=clusterName, temp_dir=workdir)   # XXX +kw
+    flags = ' !ssl'
+    ca   = kw.pop('ca',   None)
+    cert = kw.pop('cert', None)
+    key  = kw.pop('key',  None)
+    if ca or cert or key:
+        if not (ca and cert and key):
+            raise RuntimeError(sinfo("incomplete ca/cert/key provided"))
+        # neo/py does `NEOCluster.SSL = neo.test.SSL` (= (ca.crt, node.crt, node.key) )
+        flags = ' ssl'
+        NEOCluster.SSL = (ca, cert, key)
+
+    if kw:
+        raise RuntimeError(sinfo("unexpected flags: %s" % kw))
+
+    cluster = NEOCluster([clusterName], adapter='SQLite', name=clusterName, temp_dir=workdir)
     cluster.start()
     defer(cluster.stop)
 

go/neo/util.go

-// Copyright (C) 2017  Nexedi SA and Contributors.
+// Copyright (C) 2017-2020  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -21,7 +21,13 @@ package neo
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
 	"io"
+	"io/ioutil"
+
+	"lab.nexedi.com/kirr/go123/xerr"
 
 	"lab.nexedi.com/kirr/neo/go/zodb"
 	"lab.nexedi.com/kirr/neo/go/internal/log"
@@ -64,3 +70,82 @@ func before2At(before zodb.Tid) (at zodb.Tid) {
 		return zodb.TidMax
 	}
 }
+
+
+// tlsForSSL builds tls.Config from ca/cert/key files that should be interoperable with NEO/py.
+//
+// see https://lab.nexedi.com/nexedi/neoppod/blob/v1.12-61-gc1c26894/neo/lib/app.py#L74-90
+func tlsForSSL(ca, cert, key string) (_ *tls.Config, err error) {
+	defer xerr.Contextf(&err, "tls setup")
+
+	caData, err := ioutil.ReadFile(ca)
+	if err != nil {
+		return nil, err
+	}
+
+	CA := x509.NewCertPool()
+	ok := CA.AppendCertsFromPEM(caData)
+	if !ok {
+		return nil, fmt.Errorf("invalid CA")
+	}
+
+	crt, err := tls.LoadX509KeyPair(cert, key)
+	if err != nil {
+		return nil, err
+	}
+
+
+	tlsCfg := &tls.Config{
+		Certificates: []tls.Certificate{crt}, // (cert, key) as loaded
+		RootCAs: CA,                          // (ca,) as loaded
+
+		// a server also verifies cient (but also see verifyPeerCert below)
+		ClientAuth: tls.RequireAndVerifyClientCert,
+		ClientCAs: CA,
+
+		PreferServerCipherSuites: true,
+	}
+
+	// TODO only accept TLS >= 1.2 ?
+
+	// tls docs say we should parse Certificate[0] into Leaf ourselves
+	leaf, err := x509.ParseCertificate(crt.Certificate[0])
+	if err != nil {
+		return nil, err
+	}
+	crt.Leaf = leaf
+
+	// NEO/py does not verify CommonName (ssl.check_hostname=False implicitly).
+	// Match that behaviour with custom VerifyPeerCertificate because Go
+	// does not provide functionality to skip only CN verification out of the box.
+	// https://github.com/golang/go/issues/21971#issuecomment-332693931
+	// https://stackoverflow.com/questions/44295820
+	verifyPeerCert := func(rawCerts [][]byte, _ [][]*x509.Certificate) (err error) {
+		defer xerr.Contextf(&err, "verify peer cert")
+
+		certv := []*x509.Certificate{}
+		for _, certData := range rawCerts {
+			cert, err := x509.ParseCertificate(certData)
+			if err != nil {
+				return err
+			}
+			certv = append(certv, cert)
+		}
+
+		vopt := x509.VerifyOptions{
+			DNSName:       "",                // means "don't verify name"
+			Roots:         tlsCfg.RootCAs,
+			Intermediates: x509.NewCertPool(),
+		}
+		for _, cert := range certv[1:] {
+			vopt.Intermediates.AddCert(cert)
+		}
+
+		_, err = certv[0].Verify(vopt)
+		return err
+	}
+	tlsCfg.InsecureSkipVerify = true // disables all verifications including for ServerName
+	tlsCfg.VerifyPeerCertificate = verifyPeerCert
+
+	return tlsCfg, nil
+}