//go:build e2e

package rds

import (
	"fmt"
	"net"
	"strings"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/rds"
	"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
	"github.com/mulgadc/spinifex/tests/e2e/harness"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

const (
	// The table the round-trip is written into. Shared by the subtests, because a
	// row that survives a reconnection is what proves they reached one database.
	connectTable = "e2e_connect"
	connectNote  = "hello from the client VM"

	// The rotation the master-password subtest moves to. Same character rules as
	// the original.
	rotatedPassword = "e2eR0tatedSecret2"

	// PostgreSQL's enforcement parameter, under AWS's own name, and the phrase
	// its refusal carries. The engine has no server setting for this: the reject
	// rule is a pg_hba fact, and "no encryption" is how a pg_hba refusal of an
	// unencrypted connection names itself to the client.
	forceSSLParameter = "rds.force_ssl"
	plaintextRefusal  = "no encryption"

	// How long a security-group change has to take effect at the datapath. The
	// control-plane call returns as soon as the ENI is re-associated; the flow
	// tables behind it settle a moment later.
	sgSettleTimeout = 90 * time.Second
)

// TestConnectivity is the suite's client leg: a psql client inside the customer
// VPC, which is the only place a DB endpoint is reachable from, driven over SSH.
//
// Nothing here skips when the client is unavailable. A DB instance that reports
// available but accepts no connection is the failure this suite exists to catch,
// so a client VM that cannot be built fails the test rather than passing it
// quietly. The one sanctioned skip is a cluster with no base domain, where there
// is no endpoint name to resolve at all.
func TestConnectivity(t *testing.T) {
	f := requireRDSFixture(t)
	t.Parallel()
	reserveDBVMs(t, dbClass)

	suffix := time.Now().Unix()
	id := fmt.Sprintf("%s-connect-%d", dbInstancePfx, suffix)
	// An empty group, so the instance boots on the catalog's own defaults exactly
	// as one with no group of its own would. It exists only because turning
	// enforcement off is something a customer does through a group of their own.
	paramGroup := fmt.Sprintf("%s-tls-%d", dbInstancePfx, suffix)
	createParameterGroup(t, f, paramGroup)

	// Started before the client VM is built so the two boots overlap: the create
	// returns immediately and the engine bootstraps while apt runs in the client.
	harness.Phase(t, "Creating DB instance %q", id)
	createDBInstance(t, f, id, func(in *rds.CreateDBInstanceInput) {
		in.DBParameterGroupName = aws.String(paramGroup)
	})

	client := rdsClient(t, f)
	instance := waitForAvailable(t, f, id)

	endpoint := aws.StringValue(instance.Endpoint.Address)
	require.NotEmpty(t, endpoint, "an available instance must publish an endpoint")
	eni := harness.DBEndpointENI(t, f.AWS, id)
	privateIP := aws.StringValue(eni.PrivateIpAddress)
	require.NotEmpty(t, privateIP, "the endpoint ENI must carry a private address")

	byIP := harness.PSQLConn{
		Host: privateIP, Port: aws.Int64Value(instance.Endpoint.Port),
		User: dbMasterUser, Password: dbMasterPassword, DBName: dbName,
	}

	// The first assertion in the suite that proves an engine is actually serving:
	// every status before this one is the control plane's own account of itself.
	t.Run("AClientInTheVPCRoundTripsARow", func(t *testing.T) {
		harness.PSQL(t, client, byIP, fmt.Sprintf(
			"CREATE TABLE %s (id int primary key, note text); INSERT INTO %s VALUES (1, '%s');",
			connectTable, connectTable, connectNote))

		out := harness.PSQL(t, client, byIP, fmt.Sprintf("SELECT note FROM %s WHERE id = 1;", connectTable))
		assert.Equal(t, connectNote, strings.TrimSpace(out), "the row written over the endpoint must read back")
	})

	// The name is the customer's handle on the instance, and it has to resolve
	// where a customer resolves it — inside the VPC. A lookup on the runner proves
	// nothing, because the runner's resolver may point straight at northstar.
	t.Run("TheEndpointNameResolvesInTheGuest", func(t *testing.T) {
		requireEndpointName(t, endpoint)
		assert.Equal(t, fmt.Sprintf("%s.%s.%s.rds.%s", id, f.Account, f.Region, f.BaseDomain), endpoint,
			"the endpoint name is account-qualified so identifiers collide across tenants without colliding in DNS")

		addrs := harness.ResolveInGuest(t, client, endpoint)
		assert.Contains(t, addrs, privateIP, "the endpoint name must resolve to the customer ENI's address")

		byName := byIP
		byName.Host = endpoint
		out := harness.PSQL(t, client, byName, fmt.Sprintf("SELECT note FROM %s WHERE id = 1;", connectTable))
		assert.Equal(t, connectNote, strings.TrimSpace(out),
			"the name and the address must reach the same database")
	})

	// The AWS service endpoints share nothing with the RDS instance under test
	// except the guest's resolver: they are cluster-wide names published under
	// AWS.ServicesDomain, so any DB instance's client VM proves them. A lookup
	// from the runner would prove nothing here either, for the same reason as
	// the endpoint name above — the runner's resolver is not the guest's.
	t.Run("TheServiceEndpointNamesResolveInTheGuest", func(t *testing.T) {
		suffix := harness.NorthstarAWSInternalSuffix(f.Env)
		for _, svc := range config.AWSGWServiceNames {
			name := fmt.Sprintf("%s.%s.%s", svc, f.Region, suffix)
			t.Run(svc, func(t *testing.T) {
				addrs := harness.ResolveInGuest(t, client, name)
				require.NotEmpty(t, addrs, "%s must resolve to at least one gateway address", name)
				for _, addr := range addrs {
					assert.NotNil(t, net.ParseIP(addr), "%s must resolve to an IP literal, got %q", name, addr)
				}
			})
		}
	})

	// Both the vanity name and the ENI address are in the serving
	// certificate's SAN set specifically so verify-full works either way, and that
	// is the mode a customer who cares about TLS actually uses.
	t.Run("VerifyFullSucceedsByNameAndByIP", func(t *testing.T) {
		hosts := map[string]string{"ByIP": privateIP}
		if net.ParseIP(endpoint) == nil {
			hosts["ByName"] = endpoint
		}
		for name, host := range hosts {
			t.Run(name, func(t *testing.T) {
				conn := byIP
				conn.Host = host
				conn.SSLMode = "verify-full"
				conn.SSLRootCert = harness.RDSClientCACertPath

				out := harness.PSQL(t, client, conn,
					"SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid();")
				assert.Equal(t, "t", strings.TrimSpace(out),
					"a verify-full connection must report itself encrypted")
			})
		}
	})

	// The suite's only direct proof that enforcement is live. The agent's own
	// check reads pg_hba_file_rules back, which says what the engine *would* load;
	// only a connection the engine actually turns away says what it did load.
	t.Run("PlaintextIsRefusedAndTLSIsNot", func(t *testing.T) {
		assertRefusesPlaintext(t, client, byIP)

		encrypted := byIP
		encrypted.SSLMode = "require"
		out := harness.PSQL(t, client, encrypted, "SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid();")
		assert.Equal(t, "t", strings.TrimSpace(out),
			"sslmode=require satisfies enforcement without the client validating anything")
	})

	// Enforcement is the customer's to turn off, as it is on AWS, and both
	// directions land on a reload rather than a restart.
	t.Run("EnforcementFlipsWithoutARestart", func(t *testing.T) {
		plaintext := byIP
		plaintext.SSLMode = "disable"

		// Registered before the flip: a failure part-way would otherwise leave every
		// later test — and the instance's own teardown — on a database serving in
		// the clear.
		t.Cleanup(func() { setGroupParameter(t, f, paramGroup, forceSSLParameter, "1", "immediate") })

		harness.Step(t, "Turning %s off on %q", forceSSLParameter, id)
		setGroupParameter(t, f, paramGroup, forceSSLParameter, "0", "immediate")
		out := harness.PSQL(t, client, plaintext, "SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid();")
		assert.Equal(t, "f", strings.TrimSpace(out),
			"with enforcement off the same connection must be accepted, and in the clear")

		harness.Step(t, "Turning %s back on on %q", forceSSLParameter, id)
		setGroupParameter(t, f, paramGroup, forceSSLParameter, "1", "immediate")
		assertRefusesPlaintext(t, client, byIP)
	})

	// The endpoint ENI is injected into the customer's subnet, so the only thing
	// that can gate it is a security group. Nothing else in the suite proves the
	// groups a customer sets are in force on it.
	t.Run("ASecurityGroupGatesTheEndpoint", func(t *testing.T) {
		original := securityGroupIDs(instance)
		require.NotEmpty(t, original, "the instance must report the groups its ENI carries")

		vpcID := aws.StringValue(instance.DBSubnetGroup.VpcId)
		closed := harness.EnsureSG(t, f.Harness, vpcID, "rds-e2e-closed")

		harness.Step(t, "Re-associating %q onto a group that does not admit 5432", id)
		setSecurityGroups(t, f, id, []string{closed})
		// Restored inside the subtest as well as asserted below: a failure part-way
		// would otherwise leave every later test connecting through a closed group.
		t.Cleanup(func() { setSecurityGroups(t, f, id, original) })

		harness.EventuallyErr(t, func() error {
			if _, err := harness.TryPSQL(client, byIP, "SELECT 1;"); err != nil {
				return nil
			}
			return fmt.Errorf("%s still accepts connections through a group that does not admit %d", id, harness.PostgresEnginePort)
		}, sgSettleTimeout, 5*time.Second)

		harness.Step(t, "Restoring the original groups on %q", id)
		setSecurityGroups(t, f, id, original)
		harness.EventuallyErr(t, func() error {
			_, err := harness.TryPSQL(client, byIP, "SELECT 1;")
			return err
		}, sgSettleTimeout, 5*time.Second)
	})

	// The master user is administrative but not a PostgreSQL superuser. This
	// is the only leg that can prove it — the bootstrap unit test sees the SQL
	// that was sent, not what the engine then refuses.
	t.Run("TheMasterUserIsAdministrativeButNotASuperuser", func(t *testing.T) {
		out := harness.PSQL(t, client, byIP, "SELECT rolsuper FROM pg_roles WHERE rolname = current_user;")
		assert.Equal(t, "f", strings.TrimSpace(out),
			"the master user must not be a PostgreSQL superuser")

		// A superuser master is command execution as the postgres OS user inside
		// the DB VM, which is what makes this the one assertion that matters.
		out, err := harness.TryPSQL(client, byIP,
			"CREATE TEMP TABLE rce(l text); COPY rce FROM PROGRAM 'id';")
		require.Error(t, err, "COPY FROM PROGRAM must be refused: %s", out)
		// Matched on the phrase the engine's own refusal carries, rather than on
		// the privilege it names: PG16 reworded that half of the message.
		assert.Contains(t, out, "external program",
			"the engine must refuse the program COPY, not fail for some other reason")

		// The administrative capability a customer is actually owed, on the same
		// connection: create a database, create a role, install a trusted extension.
		harness.PSQL(t, client, byIP, "CREATE DATABASE e2e_master_admin;")
		t.Cleanup(func() { harness.PSQL(t, client, byIP, "DROP DATABASE IF EXISTS e2e_master_admin;") })

		harness.PSQL(t, client, byIP, "CREATE ROLE e2e_master_role LOGIN PASSWORD 'e2eR0leSecret1';")
		t.Cleanup(func() { harness.PSQL(t, client, byIP, "DROP ROLE IF EXISTS e2e_master_role;") })

		harness.PSQL(t, client, byIP, "CREATE EXTENSION IF NOT EXISTS pg_trgm;")
		out = harness.PSQL(t, client, byIP, "SELECT extname FROM pg_extension WHERE extname = 'pg_trgm';")
		assert.Equal(t, "pg_trgm", strings.TrimSpace(out),
			"the master user must be able to install a trusted extension in a database it owns")
	})

	// Last: it retires the credential every subtest above connects with. The system keeps
	// no cleartext password anywhere, so the only proof the rotation reached the
	// engine is that the new one authenticates and the old one does not.
	t.Run("TheMasterPasswordRotates", func(t *testing.T) {
		_, err := f.AWS.RDS.ModifyDBInstance(&rds.ModifyDBInstanceInput{
			DBInstanceIdentifier: aws.String(id),
			MasterUserPassword:   aws.String(rotatedPassword),
			ApplyImmediately:     aws.Bool(true),
		})
		require.NoError(t, err, "modify-db-instance master password")

		rotated := byIP
		rotated.Password = rotatedPassword
		out := harness.PSQL(t, client, rotated, fmt.Sprintf("SELECT note FROM %s WHERE id = 1;", connectTable))
		assert.Equal(t, connectNote, strings.TrimSpace(out), "the rotated password must reach the same database")

		out, err = harness.TryPSQL(client, byIP, "SELECT 1;")
		require.Error(t, err, "the retired password must not connect: %s", out)
		assert.Contains(t, out, "authentication",
			"the old password must be refused by the engine, not by the network")
	})
}

// assertRefusesPlaintext is the assertion every PostgreSQL transition repeats:
// enforcement lives on the data volume, so an instance that came back not
// enforcing is available, healthy and serving in the clear.
//
// The refusal is matched on the reason rather than the whole message, which
// differs from AWS's own wording for the same rejection.
func assertRefusesPlaintext(t *testing.T, tgt harness.SSHTarget, conn harness.PSQLConn) {
	t.Helper()
	plaintext := conn
	plaintext.SSLMode = "disable"
	out, err := harness.TryPSQL(tgt, plaintext, "SELECT 1;")
	require.Error(t, err, "a plaintext connection must be refused: %s", out)
	assert.Contains(t, out, plaintextRefusal,
		"the engine must turn the connection away for being unencrypted, not for some other reason")
}

// The suite's one sanctioned skip on the client leg: with no base domain the
// endpoint is the bare ENI address, so there is no name to resolve and nothing
// a hosts-file entry could honestly stand in for.
func requireEndpointName(t *testing.T, endpoint string) {
	t.Helper()
	if net.ParseIP(endpoint) != nil {
		t.Skipf("endpoint %s is a bare ENI address; no northstar base domain is configured", endpoint)
	}
}

func securityGroupIDs(instance *rds.DBInstance) []string {
	ids := make([]string, 0, len(instance.VpcSecurityGroups))
	for _, g := range instance.VpcSecurityGroups {
		ids = append(ids, aws.StringValue(g.VpcSecurityGroupId))
	}
	return ids
}

// Re-associates the endpoint ENI's groups. Non-disruptive, so it lands
// immediately whatever ApplyImmediately says, and the call returns before the
// datapath has caught up — hence the polls at every call site.
func setSecurityGroups(t *testing.T, f *Fixture, id string, groups []string) {
	t.Helper()
	out, err := f.AWS.RDS.ModifyDBInstance(&rds.ModifyDBInstanceInput{
		DBInstanceIdentifier: aws.String(id),
		VpcSecurityGroupIds:  aws.StringSlice(groups),
		ApplyImmediately:     aws.Bool(true),
	})
	if err != nil {
		t.Fatalf("modify-db-instance security groups %v: %v", groups, err)
	}
	assert.ElementsMatch(t, groups, securityGroupIDs(out.DBInstance),
		"the describe must report the groups the ENI now carries")
}