package backends

import (
	"testing"

	. "github.com/iegomez/mosquitto-go-auth/backends/constants"
	"github.com/iegomez/mosquitto-go-auth/hashing"
	log "github.com/sirupsen/logrus"
	. "github.com/smartystreets/goconvey/convey"
)

func TestMysql(t *testing.T) {

	//Initialize Mysql without mandatory values (fail).
	authOpts := make(map[string]string)
	authOpts["mysql_host"] = "localhost"
	authOpts["mysql_port"] = "3306"
	authOpts["mysql_protocol"] = "tcp"
	authOpts["mysql_allow_native_passwords"] = "true"

	Convey("If mandatory params are not set initialization should fail", t, func() {
		_, err := NewMysql(authOpts, log.DebugLevel, hashing.NewHasher(authOpts, "mysql"))
		So(err, ShouldBeError)
	})

	//Initialize Mysql with some test values (omit tls).
	authOpts["mysql_dbname"] = "go_auth_test"
	authOpts["mysql_user"] = "go_auth_test"
	authOpts["mysql_password"] = "go_auth_test"
	authOpts["mysql_userquery"] = "SELECT password_hash FROM test_user WHERE username = ? limit 1"
	authOpts["mysql_superquery"] = "select count(*) from test_user where username = ? and is_admin = true"
	authOpts["mysql_aclquery"] = "SELECT test_acl.topic FROM test_acl, test_user WHERE test_user.username = ? AND test_acl.test_user_id = test_user.id AND (rw >= ? or rw = 3)"

	Convey("Given valid params NewMysql should return a Mysql backend instance", t, func() {
		mysql, err := NewMysql(authOpts, log.DebugLevel, hashing.NewHasher(authOpts, "mysql"))
		So(err, ShouldBeNil)

		//Empty db
		mysql.DB.MustExec("delete from test_user where 1 = 1")
		mysql.DB.MustExec("delete from test_acl where 1 = 1")

		//Insert a user to test auth
		username := "test"
		userPass := "testpw"
		//Hash generated by the pw utility
		userPassHash := "PBKDF2$sha512$100000$os24lcPr9cJt2QDVWssblQ==$BK1BQ2wbwU1zNxv3Ml3wLuu5//hPop3/LvaPYjjCwdBvnpwusnukJPpcXQzyyjOlZdieXTx6sXAcX4WnZRZZnw=="
		wrongUsername := "not_present"

		insertQuery := "INSERT INTO test_user(username, password_hash, is_admin) values(?, ?, ?)"

		var userID int64

		res, err := mysql.DB.Exec(insertQuery, username, userPassHash, true)
		So(err, ShouldBeNil)

		userID, err = res.LastInsertId()

		So(err, ShouldBeNil)
		So(userID, ShouldBeGreaterThan, 0)

		Convey("Given a username and a correct password, it should correctly authenticate it", func() {

			authenticated, err := mysql.GetUser(username, userPass, "")
			So(err, ShouldBeNil)
			So(authenticated, ShouldBeTrue)

		})

		Convey("Given a username and an incorrect password, it should not authenticate it", func() {

			authenticated, err := mysql.GetUser(username, "wrong_password", "")
			So(err, ShouldBeNil)
			So(authenticated, ShouldBeFalse)

		})

		Convey("Given a wrong username, it should not authenticate it and not return error", func() {

			authenticated, err := mysql.GetUser(wrongUsername, "whatever_password", "")
			So(err, ShouldBeNil)
			So(authenticated, ShouldBeFalse)

		})

		Convey("Given a username that is admin, super user should pass", func() {
			superuser, err := mysql.GetSuperuser(username)
			So(err, ShouldBeNil)
			So(superuser, ShouldBeTrue)
		})

		Convey("Given a wrong username, super user should not return error", func() {
			superuser, err := mysql.GetSuperuser(wrongUsername)
			So(err, ShouldBeNil)
			So(superuser, ShouldBeFalse)
		})

		//Now create some acls and test topics

		strictAcl := "test/topic/1"
		singleLevelAcl := "test/topic/+"
		hierarchyAcl := "test/#"

		userPattern := "test/%u"
		clientPattern := "test/%c"

		clientID := "test_client"

		var aclID int64
		aclQuery := "INSERT INTO test_acl(test_user_id, topic, rw) values(?, ?, ?)"
		res, err = mysql.DB.Exec(aclQuery, userID, strictAcl, MOSQ_ACL_READ)
		So(err, ShouldBeNil)

		aclID, err = res.LastInsertId()
		So(err, ShouldBeNil)
		So(aclID, ShouldBeGreaterThan, 0)

		Convey("Given only strict acl in db, an exact match should work and and inexact one not", func() {

			testTopic1 := `test/topic/1`
			testTopic2 := `test/topic/2`

			tt1, err1 := mysql.CheckAcl(username, testTopic1, clientID, MOSQ_ACL_READ)
			tt2, err2 := mysql.CheckAcl(username, testTopic2, clientID, MOSQ_ACL_READ)

			So(err1, ShouldBeNil)
			So(err2, ShouldBeNil)
			So(tt1, ShouldBeTrue)
			So(tt2, ShouldBeFalse)

		})

		Convey("Given read only privileges, a pub check should fail", func() {

			testTopic1 := "test/topic/1"
			tt1, err1 := mysql.CheckAcl(username, testTopic1, clientID, MOSQ_ACL_WRITE)
			So(err1, ShouldBeNil)
			So(tt1, ShouldBeFalse)

		})

		Convey("Given wildcard subscriptions against strict db acl, acl checks should fail", func() {

			tt1, err1 := mysql.CheckAcl(username, singleLevelAcl, clientID, MOSQ_ACL_READ)
			tt2, err2 := mysql.CheckAcl(username, hierarchyAcl, clientID, MOSQ_ACL_READ)

			So(err1, ShouldBeNil)
			So(err2, ShouldBeNil)
			So(tt1, ShouldBeFalse)
			So(tt2, ShouldBeFalse)

		})

		//Now check against patterns.

		_, err = mysql.DB.Exec(aclQuery, userID, userPattern, MOSQ_ACL_READ)
		So(err, ShouldBeNil)

		Convey("Given a topic that mentions username, acl check should pass", func() {
			tt1, err1 := mysql.CheckAcl(username, "test/test", clientID, MOSQ_ACL_READ)
			So(err1, ShouldBeNil)
			So(tt1, ShouldBeTrue)
		})

		_, err = mysql.DB.Exec(aclQuery, userID, clientPattern, MOSQ_ACL_READ)
		So(err, ShouldBeNil)

		Convey("Given a topic that mentions clientid, acl check should pass", func() {
			tt1, err1 := mysql.CheckAcl(username, "test/test_client", clientID, MOSQ_ACL_READ)
			So(err1, ShouldBeNil)
			So(tt1, ShouldBeTrue)
		})

		//Now insert single level topic to check against.

		_, err = mysql.DB.Exec(aclQuery, userID, singleLevelAcl, MOSQ_ACL_READ)
		So(err, ShouldBeNil)

		Convey("Given a topic not strictly present that matches a db single level wildcard, acl check should pass", func() {
			tt1, err1 := mysql.CheckAcl(username, "test/topic/whatever", clientID, MOSQ_ACL_READ)
			So(err1, ShouldBeNil)
			So(tt1, ShouldBeTrue)
		})

		//Now insert hierarchy wildcard to check against.

		_, err = mysql.DB.Exec(aclQuery, userID, hierarchyAcl, MOSQ_ACL_READ)
		So(err, ShouldBeNil)

		Convey("Given a topic not strictly present that matches a hierarchy wildcard, acl check should pass", func() {
			tt1, err1 := mysql.CheckAcl(username, "test/what/ever", clientID, MOSQ_ACL_READ)
			So(err1, ShouldBeNil)
			So(tt1, ShouldBeTrue)
		})

		Convey("Given a bad username, acl check should not return error", func() {
			testTopic1 := `test/topic/1`
			tt1, err1 := mysql.CheckAcl(wrongUsername, testTopic1, clientID, MOSQ_ACL_READ)
			So(err1, ShouldBeNil)
			So(tt1, ShouldBeFalse)
		})

		//Empty db
		mysql.DB.MustExec("delete from test_user where 1 = 1")
		mysql.DB.MustExec("delete from test_acl where 1 = 1")

		mysql.Halt()

	})

}

func TestMysqlTls(t *testing.T) {
	authOpts := make(map[string]string)
	authOpts["mysql_host"] = "localhost"
	authOpts["mysql_port"] = "3306"
	authOpts["mysql_protocol"] = "tcp"
	authOpts["mysql_allow_native_passwords"] = "true"
	authOpts["mysql_dbname"] = "go_auth_test"
	authOpts["mysql_user"] = "go_auth_test_tls"
	authOpts["mysql_password"] = "go_auth_test_tls"

	authOpts["mysql_userquery"] = "SELECT password_hash FROM test_user WHERE username = ? limit 1"
	authOpts["mysql_superquery"] = "select count(*) from test_user where username = ? and is_admin = true"
	authOpts["mysql_aclquery"] = "SELECT test_acl.topic FROM test_acl, test_user WHERE test_user.username = ? AND test_acl.test_user_id = test_user.id AND (rw >= ? or rw = 3)"

	Convey("Given custom ssl disabled, it should fail", t, func() {
		mysql, err := NewMysql(authOpts, log.DebugLevel, hashing.NewHasher(authOpts, "mysql"))
		So(err, ShouldBeError)
		So(err.Error(), ShouldContainSubstring, "Access denied for user")
		So(mysql.DB, ShouldBeNil)
	})

	authOpts["mysql_sslmode"] = "custom"
	authOpts["mysql_sslrootcert"] = "/test-files/certificates/ca.pem"

	Convey("Given custom ssl enabled, it should work without a client certificate", t, func() {
		mysql, err := NewMysql(authOpts, log.DebugLevel, hashing.NewHasher(authOpts, "mysql"))
		So(err, ShouldBeNil)

		rows, err := mysql.DB.Query("SHOW status like 'Ssl_cipher';")
		So(err, ShouldBeNil)
		So(rows.Next(), ShouldBeTrue)

		var variableName string
		var variableValue string
		err = rows.Scan(&variableName, &variableValue)
		So(err, ShouldBeNil)

		So(variableName, ShouldEqual, "Ssl_cipher")
		So(variableValue, ShouldNotBeBlank)
	})
}

func TestMysqlMutualTls(t *testing.T) {
	authOpts := make(map[string]string)
	authOpts["mysql_host"] = "localhost"
	authOpts["mysql_port"] = "3306"
	authOpts["mysql_protocol"] = "tcp"
	authOpts["mysql_allow_native_passwords"] = "true"
	authOpts["mysql_dbname"] = "go_auth_test"
	authOpts["mysql_user"] = "go_auth_test_mutual_tls"
	authOpts["mysql_password"] = "go_auth_test_mutual_tls"

	authOpts["mysql_userquery"] = "SELECT password_hash FROM test_user WHERE username = ? limit 1"
	authOpts["mysql_superquery"] = "select count(*) from test_user where username = ? and is_admin = true"
	authOpts["mysql_aclquery"] = "SELECT test_acl.topic FROM test_acl, test_user WHERE test_user.username = ? AND test_acl.test_user_id = test_user.id AND (rw >= ? or rw = 3)"

	authOpts["mysql_sslmode"] = "custom"
	authOpts["mysql_sslrootcert"] = "/test-files/certificates/ca.pem"

	Convey("Given custom ssl enabled and no client certificate is given, it should fail", t, func() {
		mysql, err := NewMysql(authOpts, log.DebugLevel, hashing.NewHasher(authOpts, "mysql"))
		So(err, ShouldBeError)
		So(err.Error(), ShouldContainSubstring, "Access denied for user")
		So(mysql.DB, ShouldBeNil)
	})

	authOpts["mysql_sslcert"] = "/test-files/certificates/db/unauthorized-second-client.pem"
	authOpts["mysql_sslkey"] = "/test-files/certificates/db/unauthorized-second-client-key.pem"

	Convey("Given custom ssl enabled and unauthorized client certificate is given, it should fail", t, func() {
		mysql, err := NewMysql(authOpts, log.DebugLevel, hashing.NewHasher(authOpts, "mysql"))
		So(err, ShouldBeError)
		So(err.Error(), ShouldContainSubstring, "Access denied for user")
		So(mysql.DB, ShouldBeNil)
	})

	authOpts["mysql_sslcert"] = "/test-files/certificates/grpc/client.pem"
	authOpts["mysql_sslkey"] = "/test-files/certificates/grpc/client-key.pem"

	Convey("Given custom ssl enabled and invalid client certificate is given, it should fail", t, func() {
		mysql, err := NewMysql(authOpts, log.DebugLevel, hashing.NewHasher(authOpts, "mysql"))
		So(err, ShouldBeError)
		So(err.Error(), ShouldContainSubstring, "invalid connection")
		So(mysql.DB, ShouldBeNil)
	})

	authOpts["mysql_sslcert"] = "/test-files/certificates/db/client.pem"
	authOpts["mysql_sslkey"] = "/test-files/certificates/db/client-key.pem"

	Convey("Given custom ssl enabled and client certificate is given, it should work", t, func() {
		mysql, err := NewMysql(authOpts, log.DebugLevel, hashing.NewHasher(authOpts, "mysql"))
		So(err, ShouldBeNil)

		rows, err := mysql.DB.Query("SHOW status like 'Ssl_cipher';")
		So(err, ShouldBeNil)
		So(rows.Next(), ShouldBeTrue)

		var variableName string
		var variableValue string
		err = rows.Scan(&variableName, &variableValue)
		So(err, ShouldBeNil)

		So(variableName, ShouldEqual, "Ssl_cipher")
		So(variableValue, ShouldNotBeBlank)
	})
}