mydumper: verify file routing config (pingcap#470)

* fix file routing

* remove useless line

* remove redundant if check

lightning/config/config.go

@@ -479,6 +479,21 @@ func (cfg *Config) Adjust() error {
 		}
 	}
 
+	// adjust file routing
+	for _, rule := range cfg.Mydumper.FileRouters {
+		if filepath.IsAbs(rule.Path) {
+			relPath, err := filepath.Rel(cfg.Mydumper.SourceDir, rule.Path)
+			if err != nil {
+				return errors.Trace(err)
+			}
+			// ".." means that this path is not in source dir, so we should return an error
+			if strings.HasPrefix(relPath, "..") {
+				return errors.Errorf("file route path '%s' is not in source dir '%s'", rule.Path, cfg.Mydumper.SourceDir)
+			}
+			rule.Path = relPath
+		}
+	}
+
 	// enable default file route rule if no rules are set
 	if len(cfg.Mydumper.FileRouters) == 0 {
 		cfg.Mydumper.DefaultFileRules = true

lightning/config/config_test.go

@@ -132,6 +132,25 @@ func (s *configTestSuite) TestAdjustInvalidBackend(c *C) {
 	c.Assert(err, ErrorMatches, "invalid config: unsupported `tikv-importer\\.backend` \\(no_such_backend\\)")
 }
 
+func (s *configTestSuite) TestAdjustFileRoutePath(c *C) {
+	cfg := config.NewConfig()
+	assignMinimalLegalValue(cfg)
+
+	tmpDir := c.MkDir()
+	cfg.Mydumper.SourceDir = tmpDir
+	invalidPath := filepath.Join(tmpDir, "../test123/1.sql")
+	rule := &config.FileRouteRule{Path: invalidPath, Type: "sql", Schema: "test", Table: "tbl"}
+	cfg.Mydumper.FileRouters = []*config.FileRouteRule{rule}
+	err := cfg.Adjust()
+	c.Assert(err, ErrorMatches, fmt.Sprintf("file route path '%s' is not in source dir '%s'", invalidPath, tmpDir))
+
+	relPath := "test_dir/1.sql"
+	rule.Path = filepath.Join(tmpDir, relPath)
+	err = cfg.Adjust()
+	c.Assert(err, IsNil)
+	c.Assert(cfg.Mydumper.FileRouters[0].Path, Equals, relPath)
+}
+
 func (s *configTestSuite) TestDecodeError(c *C) {
 	ts, host, port := startMockServer(c, http.StatusOK, "invalid-string")
 	defer ts.Close()

lightning/mydump/router.go

@@ -176,6 +176,9 @@ type regexRouterParser struct{}
 
 func (p regexRouterParser) Parse(r *config.FileRouteRule) (*RegexRouter, error) {
 	rule := &RegexRouter{}
+	if r.Path == "" && r.Pattern == "" {
+		return nil, errors.New("`path` and `pattern` must not be both empty in [[mydumper.files]]")
+	}
 	if r.Path != "" && r.Pattern != "" {
 		return nil, errors.New("can't set both `path` and `pattern` field in [[mydumper.files]]")
 	}

lightning/mydump/router_test.go

@@ -40,6 +40,37 @@ func (t *testFileRouterSuite) TestRouteParser(c *C) {
 	}
 }
 
+func (t *testFileRouterSuite) TestInvalidRouteRule(c *C) {
+	rule := &config.FileRouteRule{}
+	rules := []*config.FileRouteRule{rule}
+	_, err := NewFileRouter(rules)
+	c.Assert(err, ErrorMatches, "`path` and `pattern` must not be both empty in \\[\\[mydumper.files\\]\\]")
+
+	rule.Pattern = `^(?:[^/]*/)*([^/.]+)\.(?P<table>[^./]+)(?:\.(?P<key>[0-9]+))?\.(?P<type>csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`
+	_, err = NewFileRouter(rules)
+	c.Assert(err, ErrorMatches, "field 'type' match pattern can't be empty")
+
+	rule.Type = "$type"
+	_, err = NewFileRouter(rules)
+	c.Assert(err, ErrorMatches, "field 'schema' match pattern can't be empty")
+
+	rule.Schema = "$schema"
+	_, err = NewFileRouter(rules)
+	c.Assert(err, ErrorMatches, "invalid named capture '\\$schema'")
+
+	rule.Schema = "$1"
+	_, err = NewFileRouter(rules)
+	c.Assert(err, ErrorMatches, "field 'table' match pattern can't be empty")
+
+	rule.Table = "$table"
+	_, err = NewFileRouter(rules)
+	c.Assert(err, IsNil)
+
+	rule.Path = "/tmp/1.sql"
+	_, err = NewFileRouter(rules)
+	c.Assert(err, ErrorMatches, "can't set both `path` and `pattern` field in \\[\\[mydumper.files\\]\\]")
+}
+
 func (t *testFileRouterSuite) TestSingleRouteRule(c *C) {
 	rules := []*config.FileRouteRule{
 		{Pattern: `^(?:[^/]*/)*([^/.]+)\.(?P<table>[^./]+)(?:\.(?P<key>[0-9]+))?\.(?P<type>csv|sql)(?:\.(?P<cp>[A-Za-z0-9]+))?$`, Schema: "$1", Table: "$table", Type: "$type", Key: "$key", Compression: "$cp"},