diff --git a/cmd/cloud_sql_proxy/cloud_sql_proxy.go b/cmd/cloud_sql_proxy/cloud_sql_proxy.go index 71d788c73..269e75006 100644 --- a/cmd/cloud_sql_proxy/cloud_sql_proxy.go +++ b/cmd/cloud_sql_proxy/cloud_sql_proxy.go @@ -84,11 +84,13 @@ func main() { return } - instances := strings.Split(*instances, ",") - if len(instances) == 1 && instances[0] == "" { - instances = nil + onGCE := onGCE() + if !onGCE && *instanceSrc != "" { + log.Fatal("-instances_metadata unsupported outside of Google Compute Engine") } - if err := Check(*dir, *useFuse, instances, *instanceSrc); err != nil { + + cfgs, err := CreateInstanceConfigs(*dir, *useFuse, strings.Split(*instances, ","), *instanceSrc) + if err != nil { log.Fatal(err) } @@ -123,7 +125,7 @@ func main() { }() } - c, err := WatchInstances(*dir, instances, updates) + c, err := WatchInstances(*dir, cfgs, updates) if err != nil { log.Fatal(err) } @@ -146,14 +148,16 @@ func main() { log.Fatalf("invalid json file %q: %v", file, err) } client = auth.NewClientFrom(cfg.TokenSource(context.Background())) - } else if *token != "" || onGCE() { + } else if *token != "" || onGCE { // Passing token == "" causes the GCE metadata server to be used. client = auth.NewAuthenticatedClient(*token) } else { log.Fatal("No authentication method available! When not running on Google Compute Engine, provide the -credential_file flag.") } - log.Print("Socket prefix: " + *dir) + if *dir != "" { + log.Print("Socket prefix: " + *dir) + } src, err := certs.NewCertSource(*host, client, *checkRegion) if err != nil { diff --git a/cmd/cloud_sql_proxy/proxy.go b/cmd/cloud_sql_proxy/proxy.go index 8a52e57ba..0c3937a1a 100644 --- a/cmd/cloud_sql_proxy/proxy.go +++ b/cmd/cloud_sql_proxy/proxy.go @@ -17,14 +17,17 @@ package main // This file contains code for supporting local sockets for the Cloud SQL Proxy. import ( + "bytes" "errors" "fmt" "log" "net" "os" "path/filepath" + "runtime" "strings" + "github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/fuse" "github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/proxy" ) @@ -32,40 +35,39 @@ import ( // local connections. Values received from the updates channel are // interpretted as a comma-separated list of instances. The set of sockets in // 'dir' is the union of 'instances' and the most recent list from 'updates'. -func WatchInstances(dir string, instances []string, updates <-chan string) (<-chan proxy.Conn, error) { +func WatchInstances(dir string, cfgs []instanceConfig, updates <-chan string) (<-chan proxy.Conn, error) { ch := make(chan proxy.Conn, 1) // Instances specified statically (e.g. as flags to the binary) will always // be available. They are ignored if also returned by the GCE metadata since // the socket will already be open. - staticInstances := make(map[string]net.Listener, len(instances)) - for _, v := range instances { - if v = strings.TrimSpace(v); v == "" { - continue - } - l, err := listenInstance(ch, dir, v) + staticInstances := make(map[string]net.Listener, len(cfgs)) + for _, v := range cfgs { + l, err := listenInstance(ch, v) if err != nil { return nil, err } - staticInstances[v] = l + staticInstances[v.Instance] = l } if updates != nil { - go watchInstancesLoop(ch, dir, updates, staticInstances) + go watchInstancesLoop(dir, ch, updates, staticInstances) } return ch, nil } -func watchInstancesLoop(dst chan<- proxy.Conn, dir string, updates <-chan string, static map[string]net.Listener) { +func watchInstancesLoop(dir string, dst chan<- proxy.Conn, updates <-chan string, static map[string]net.Listener) { dynamicInstances := make(map[string]net.Listener) for instances := range updates { + list, err := parseInstanceConfigs(dir, strings.Split(instances, ",")) + if err != nil { + log.Print(err) + } + stillOpen := make(map[string]net.Listener) + for _, cfg := range list { + instance := cfg.Instance - list := strings.Split(instances, ",") - for _, instance := range list { - if len(instance) == 0 { - continue - } // If the instance is specified in the static list don't do anything: // it's already open and should stay open forever. if _, ok := static[instance]; ok { @@ -78,7 +80,7 @@ func watchInstancesLoop(dst chan<- proxy.Conn, dir string, updates <-chan string continue } - l, err := listenInstance(dst, dir, instance) + l, err := listenInstance(dst, cfg) if err != nil { log.Printf("Couldn't open socket for %q: %v", instance, err) continue @@ -116,33 +118,18 @@ func remove(path string) { // listenInstance starts listening on a new unix socket in dir to connect to the // specified instance. New connections to this socket are sent to dst. -func listenInstance(dst chan<- proxy.Conn, dir, instance string) (net.Listener, error) { - log.Printf("listenInstance: %q", instance) - - var path string - var l net.Listener - if eq := strings.Index(instance, "="); eq != -1 { - spl := strings.SplitN(instance[eq+1:], ":", 2) - if len(spl) == 1 { - return nil, fmt.Errorf("invalid format in %q; expected 'project:instance=tcp:port'", instance) - } - - instance = instance[:eq] - - var err error - if l, err = net.Listen(spl[0], "127.0.0.1:"+spl[1]); err != nil { - return nil, err - } - path = "localhost:" + spl[1] - } else { - path = filepath.Join(dir, instance) - remove(path) - var err error - if l, err = net.Listen("unix", path); err != nil { - return nil, err - } - if err := os.Chmod(path, 0777|os.ModeSocket); err != nil { - log.Printf("couldn't update permissions for socket file %q: %v; other users may not be unable to connect", path, err) +func listenInstance(dst chan<- proxy.Conn, cfg instanceConfig) (net.Listener, error) { + unix := cfg.Network == "unix" + if unix { + remove(cfg.Address) + } + l, err := net.Listen(cfg.Network, cfg.Address) + if err != nil { + return nil, err + } + if unix { + if err := os.Chmod(cfg.Address, 0777|os.ModeSocket); err != nil { + log.Printf("couldn't update permissions for socket file %q: %v; other users may not be unable to connect", cfg.Address, err) } } @@ -150,31 +137,176 @@ func listenInstance(dst chan<- proxy.Conn, dir, instance string) (net.Listener, for { c, err := l.Accept() if err != nil { - log.Printf("Error in accept for %q on %v: %v", instance, path, err) + log.Printf("Error in accept for %q on %v: %v", cfg, cfg.Address, err) l.Close() return } - log.Printf("Got a connection for %q", instance) - dst <- proxy.Conn{instance, c} + log.Printf("Got a connection for %q", cfg.Instance) + dst <- proxy.Conn{cfg.Instance, c} } }() - log.Printf("Open socket for %q at %q", instance, path) + log.Printf("Open socket for %q at %q", cfg.Instance, cfg.Address) return l, nil } -// Check verifies that the dir parameter is set and that either 'fuse' is true or -// at least one of instances/instancesSrc is set, but not both. -func Check(dir string, fuse bool, instances []string, instancesSrc string) error { - switch { - case dir == "": - return errors.New("must set -dir") - case !fuse: - if len(instances) == 0 && instancesSrc == "" { - return errors.New("must specify -fuse, -instances, or -instances_metadata") +type instanceConfig struct { + Instance string + Network, Address string +} + +// loopbackForNet maps a network (e.g. tcp6) to the loopback address for that +// network. It is updated during the initialization of validNets to include a +// valid loopback address for "tcp". +var loopbackForNet = map[string]string{ + "tcp4": "127.0.0.1", + "tcp6": "[::1]", +} + +// validNets tracks the networks that are valid for this platform and machine. +var validNets = func() map[string]bool { + m := map[string]bool{ + "unix": runtime.GOOS != "windows", + } + + anyTCP := false + for _, n := range []string{"tcp4", "tcp6"} { + addr, ok := loopbackForNet[n] + if !ok { + // This is effectively a compile-time error. + panic(fmt.Sprintf("no loopback address found for %v", n)) + } + // Open any port to see if the net is valid. + x, err := net.Listen(n, addr+":") + if err != nil { + log.Printf("Protocol %v not supported: %v", n, err) + continue + } + x.Close() + m[n] = true + + if !anyTCP { + anyTCP = true + // Set the loopback value for generic tcp if it hasn't already been + // set. (If both tcp4/tcp6 are supported the first one in the list + // (tcp4's 127.0.0.1) is used. + loopbackForNet["tcp"] = addr + } + } + if anyTCP { + m["tcp"] = true + } + return m +}() + +func parseInstanceConfig(dir, instance string) (instanceConfig, error) { + var ret instanceConfig + eq := strings.Index(instance, "=") + if eq != -1 { + spl := strings.SplitN(instance[eq+1:], ":", 3) + ret.Instance = instance[:eq] + + switch len(spl) { + default: + return ret, fmt.Errorf("invalid %q: expected 'project:instance=tcp:port'", instance) + case 2: + // No "host" part of the address. Be safe and assume that they want a + // loopback address. + ret.Network = spl[0] + addr, ok := loopbackForNet[spl[0]] + if !ok { + return ret, fmt.Errorf("invalid %q: unrecognized network %v", instance, spl[0]) + } + ret.Address = fmt.Sprintf("%s:%s", addr, spl[1]) + case 3: + // User provided a host and port; use that. + ret.Network = spl[0] + ret.Address = fmt.Sprintf("%s:%s", spl[1], spl[2]) + } + } else { + ret.Instance = instance + // Default to unix socket. + ret.Network = "unix" + ret.Address = filepath.Join(dir, instance) + } + + if !validNets[ret.Network] { + return ret, fmt.Errorf("invalid %q: unsupported network: %v", instance, ret.Network) + } + return ret, nil +} + +// parseInstanceConfigs calls parseInstanceConfig for each instance in the +// provided slice, collecting errors along the way. There may be valid +// instanceConfigs returned even if there's an error. +func parseInstanceConfigs(dir string, instances []string) ([]instanceConfig, error) { + errs := new(bytes.Buffer) + var cfg []instanceConfig + for _, v := range instances { + if v == "" { + continue + } + if c, err := parseInstanceConfig(dir, v); err != nil { + fmt.Fprintf(errs, "\n\t%v", err) + } else { + cfg = append(cfg, c) + } + } + + var err error + if errs.Len() > 0 { + err = fmt.Errorf("errors parsing config:%s", errs) + } + return cfg, err +} + +// CreateInstanceConfigs verifies that the parameters passed to it are valid +// for the proxy for the platform and system and then returns a slice of valid +// instanceConfig. +func CreateInstanceConfigs(dir string, useFuse bool, instances []string, instancesSrc string) ([]instanceConfig, error) { + if len(instances) == 1 && instances[0] == "" { + instances = nil + } + if useFuse && !fuse.Supported() { + return nil, errors.New("FUSE not supported on this system") + } + + cfgs, err := parseInstanceConfigs(dir, instances) + if err != nil { + return nil, err + } + + if dir == "" { + // Reasons to set '-dir': + // - Using -fuse + // - Using the metadata to get a list of instances + // - Having an instance that uses a 'unix' network + if useFuse { + return nil, errors.New("must set -dir because -fuse was set") + } else if instancesSrc != "" { + return nil, errors.New("must set -dir because -instances_metadata was set") + } else { + for _, v := range cfgs { + if v.Network == "unix" { + return nil, fmt.Errorf("must set -dir: using a unix socket for %v", v.Instance) + } + } + } + // Otherwise it's safe to not set -dir + } + + if useFuse { + if len(instances) != 0 || instancesSrc != "" { + return nil, errors.New("-fuse is not compatible with -instances or -instances_metadata") + } + return nil, nil + } + // FUSE disabled. + if len(instances) == 0 && instancesSrc == "" { + if fuse.Supported() { + return nil, errors.New("must specify -fuse, -instances, or -instances_metadata") } - case len(instances) != 0 || instancesSrc != "": - return errors.New("-fuse is not compatible with -instances or -instances_metadata") + return nil, errors.New("must specify -instances") } - return nil + return cfgs, nil } diff --git a/cmd/cloud_sql_proxy/proxy_test.go b/cmd/cloud_sql_proxy/proxy_test.go new file mode 100644 index 000000000..78ce11ad1 --- /dev/null +++ b/cmd/cloud_sql_proxy/proxy_test.go @@ -0,0 +1,120 @@ +package main + +import ( + "testing" +) + +func TestCreateInstanceConfigs(t *testing.T) { + for _, v := range []struct { + desc string + //inputs + dir string + useFuse bool + instances []string + instancesSrc string + + // We don't need to check the []instancesConfig return value, we already + // have a TestParseInstanceConfig. + wantErr bool + }{ + { + "setting -fuse and -dir", + "dir", true, nil, "", false, + }, { + "setting -fuse", + "", true, nil, "", true, + }, { + "setting -fuse, -dir, and -instances", + "dir", true, []string{"x"}, "", true, + }, { + "setting -fuse, -dir, and -instances_metadata", + "dir", true, nil, "md", true, + }, { + "setting -dir and -instances (unix socket)", + "dir", false, []string{"x"}, "", false, + }, { + "Seting -instance (unix socket)", + "", false, []string{"x"}, "", true, + }, { + "setting -instance (tcp socket)", + "", false, []string{"x=tcp:1234"}, "", false, + }, { + "setting -instance (tcp socket) and -instances_metadata", + "", false, []string{"x=tcp:1234"}, "md", true, + }, { + "setting -dir, -instance (tcp socket), and -instances_metadata", + "dir", false, []string{"x=tcp:1234"}, "md", false, + }, { + "setting -dir, -instance (unix socket), and -instances_metadata", + "dir", false, []string{"x"}, "md", false, + }, { + "setting -dir and -instances_metadata", + "dir", false, nil, "md", false, + }, { + "setting -instances_metadata", + "", false, nil, "md", true, + }, + } { + _, err := CreateInstanceConfigs(v.dir, v.useFuse, v.instances, v.instancesSrc) + if v.wantErr { + if err == nil { + t.Errorf("CreateInstanceConfigs passed when %s, wanted error", v.desc) + } + continue + } + if err != nil { + t.Errorf("CreateInstanceConfigs gave error when %s: %v", v.desc, err) + } + } +} + +func TestParseInstanceConfig(t *testing.T) { + for _, v := range []struct { + // inputs + dir, instance string + + wantCfg instanceConfig + wantErr bool + }{ + { + "/x", "my-instance", + instanceConfig{"my-instance", "unix", "/x/my-instance"}, + false, + }, { + "/x", "my-instance=tcp:1234", + instanceConfig{"my-instance", "tcp", "127.0.0.1:1234"}, + false, + }, { + "/x", "my-instance=tcp:my-host:1111", + instanceConfig{"my-instance", "tcp", "my-host:1111"}, + false, + }, { + "/x", "my-instance=", + instanceConfig{}, + true, + }, { + "/x", "my-instance=cool network", + instanceConfig{}, + true, + }, { + "/x", "my-instance=cool network:1234", + instanceConfig{}, + true, + }, { + "/x", "my-instance=oh:so:many:colons", + instanceConfig{}, + true, + }, + } { + got, err := parseInstanceConfig(v.dir, v.instance) + if v.wantErr { + if err == nil { + t.Errorf("parseInstanceConfig(%s, %s) = %+v, wanted error", got) + } + continue + } + if got != v.wantCfg { + t.Errorf("parseInstanceConfig(%s, %s) = %+v, want %+v", v.dir, v.instance, got, v.wantCfg) + } + } +} diff --git a/proxy/fuse/fuse.go b/proxy/fuse/fuse.go index 7e4ebbcb3..397511e2a 100644 --- a/proxy/fuse/fuse.go +++ b/proxy/fuse/fuse.go @@ -50,6 +50,12 @@ import ( "golang.org/x/net/context" ) +// Supported returns true if the current system supports FUSE. +// TODO: for OSX, check to see if OSX FUSE is installed. +func Supported() bool { + return true +} + // NewConnSrc returns a source of new connections based on Lookups in the // provided mount directory. If there isn't a directory located at tmpdir one // is created. The second return parameter can be used to shutdown and release diff --git a/proxy/fuse/fuse_windows.go b/proxy/fuse/fuse_windows.go index 253031ccb..3f49c0b14 100644 --- a/proxy/fuse/fuse_windows.go +++ b/proxy/fuse/fuse_windows.go @@ -22,6 +22,10 @@ import ( "github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/proxy" ) +func Supported() bool { + return false +} + func NewConnSrc(mountdir, tmpdir string, connset *proxy.ConnSet) (<-chan proxy.Conn, io.Closer, error) { return nil, nil, errors.New("fuse not supported on windows") }