package main

// The store's reader reads only what a module declares readable, never a secret (novox/hq issue 337).
// What the server does with the grants is proven against a real one (live_reader_test.go); here, that
// the module asks for the right ones.

import (
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
)

func TestReadableFromAContribution(t *testing.T) {
	got, err := ParseReadable(map[string]any{"name": "x", "readable": []any{"agents", "audit.events", "steps( id , created_at )"}})
	if err != nil {
		t.Fatal(err)
	}
	want := []string{"public.agents", "audit.events", "public.steps(id, created_at)"}
	if len(got) != len(want) {
		t.Fatalf("%v", got)
	}
	for i := range want {
		if got[i].String() != want[i] {
			t.Fatalf("%d: %s, want %s", i, got[i], want[i])
		}
	}
	if got[0].Columns != nil {
		t.Fatal("a bare table is every column, not none")
	}
	if none, err := ParseReadable(map[string]any{"name": "x"}); err != nil || none != nil {
		t.Fatal("absent is nothing readable:", none, err)
	}
	for _, bad := range []any{
		"agents",                         // not a list
		[]any{"agents; drop table x"},    // not a name
		[]any{"steps(id, created_at"},    // unbalanced
		[]any{"steps(id, \"x\")"},        // a quoted column
		[]any{"agents", "public.agents"}, // twice
		[]any{42},                        // not a string
		[]any{"a.b.c"},                   // three parts
	} {
		if _, err := ParseReadable(map[string]any{"readable": bad}); err == nil {
			t.Errorf("%v was accepted", bad)
		}
	}
}

func TestSecretNames(t *testing.T) {
	for _, secret := range []string{"private", "signing_key", "sealed", "operator_sealed", "module_secret", "secret",
		"operator_key", "node_key", "overlay_key", "sealing_key", "password_hash", "passwd", "salt", "rands", "token",
		"refresh_token", "secret_data", "credential_data", "api_key", "apikey", "totp_secret", "otp", "key", "client_secret"} {
		if !SecretName(secret) {
			t.Errorf("%s is not taken for a secret", secret)
		}
	}
	for _, plain := range []string{"name", "created", "commit_hash", "status", "node", "keyboard", "monkey", "footprint",
		"description", "id", "retired", "public"} {
		if SecretName(plain) {
			t.Errorf("%s is taken for a secret", plain)
		}
	}
}

func TestTheReaderIsGivenOnlyWhatIsDeclaredAndNeverASecret(t *testing.T) {
	decl, _ := ParseReadable(map[string]any{"readable": []any{"node", "signing_key", "build(id, sealed, nope)", "gone"}})
	existing := []column{
		{"public", "node", "name"}, {"public", "node", "overlay_key"}, {"public", "node", "role"},
		{"public", "signing_key", "public"}, {"public", "signing_key", "private"},
		{"public", "build", "id"}, {"public", "build", "sealed"}, {"public", "build", "log"},
	}
	stmts, rep := planDatabase(decl, existing, newHeld(), map[string]bool{})
	sql := strings.Join(stmts, "\n")
	if !strings.Contains(sql, `GRANT SELECT ("name", "role") ON "public"."node" TO "mesh_store_reader"`) ||
		!strings.Contains(sql, `GRANT SELECT ("id") ON "public"."build" TO "mesh_store_reader"`) {
		t.Fatal(sql)
	}
	for _, never := range []string{"overlay_key", "private", "sealed", "signing_key", `"log"`} {
		if strings.Contains(sql, never) {
			t.Fatalf("%s granted:\n%s", never, sql)
		}
	}
	if len(rep.Refused) != 3 || len(rep.Missing) != 2 {
		t.Fatalf("refused %v, missing %v", rep.Refused, rep.Missing)
	}
	if strings.Join(rep.Readable, ",") != "public.build.id,public.node.name,public.node.role" {
		t.Fatal(rep.Readable)
	}
}

func TestWhatTheReaderHoldsAndIsNotDeclaredIsRevoked(t *testing.T) {
	// An install before issue 337, or a grant by hand: a table-level grant, a column no longer
	// declared, and a declared column held with more than SELECT.
	sel := func(by string) grant { return grant{"SELECT", by} }
	h := held{
		Tables: map[table]tableGrants{
			{"public", "secret"}: {Kind: "r", Grants: []grant{sel("postgres")}},
			{"public", "seq"}:    {Kind: "S", Grants: []grant{{"USAGE", "postgres"}}},
		},
		Columns: map[column][]grant{
			{"public", "node", "name"}: {sel("postgres")},
			{"public", "node", "role"}: {sel("postgres"), {"UPDATE", "postgres"}},
			{"public", "node", "old"}:  {sel("postgres")},
		},
	}
	decl, _ := ParseReadable(map[string]any{"readable": []any{"node(name, role)"}})
	existing := []column{{"public", "node", "name"}, {"public", "node", "role"}, {"public", "node", "old"}}
	stmts, rep := planDatabase(decl, existing, h, map[string]bool{})
	want := []string{
		`SET ROLE "postgres"`,
		`REVOKE ALL ("old", "role") ON "public"."node" FROM "mesh_store_reader"`,
		`REVOKE ALL ON SEQUENCE "public"."seq" FROM "mesh_store_reader"`,
		`REVOKE ALL ON TABLE "public"."secret" FROM "mesh_store_reader"`,
		`RESET ROLE`,
		`GRANT SELECT ("name", "role") ON "public"."node" TO "mesh_store_reader"`,
	}
	if strings.Join(stmts, "\n") != strings.Join(want, "\n") {
		t.Fatalf("got\n%s\nwant\n%s", strings.Join(stmts, "\n"), strings.Join(want, "\n"))
	}
	if len(rep.Revoked) != 4 {
		t.Fatal(rep.Revoked)
	}
	// Undeclared: everything held goes, and nothing is granted.
	stmts, _ = planDatabase(nil, nil, h, map[string]bool{})
	for _, s := range stmts {
		if strings.HasPrefix(s, "GRANT") {
			t.Fatal(s)
		}
	}
	if len(stmts) != 5 {
		t.Fatal(stmts)
	}
	// Held exactly as declared: nothing to do.
	exact := held{Tables: map[table]tableGrants{}, Columns: map[column][]grant{{"public", "node", "name"}: {sel("postgres")}}}
	decl, _ = ParseReadable(map[string]any{"readable": []any{"node(name)"}})
	if stmts, _ := planDatabase(decl, existing, exact, map[string]bool{}); len(stmts) != 0 {
		t.Fatal(stmts)
	}
}

// A superuser's REVOKE acts as the table's owner, so a grant to the reader made by another role
// holding GRANT OPTION survives it. Each grant is revoked as its own grantor.
func TestAGrantIsRevokedAsItsOwnGrantor(t *testing.T) {
	h := held{
		Tables:  map[table]tableGrants{{"public", "vault"}: {Kind: "r", Grants: []grant{{"SELECT", "middle"}, {"SELECT", "owner1"}}}},
		Columns: map[column][]grant{{"public", "vault", "name"}: {{"SELECT", "middle"}}},
	}
	stmts, _ := planDatabase(nil, nil, h, map[string]bool{})
	want := []string{
		`SET ROLE "middle"`,
		`REVOKE ALL ("name") ON "public"."vault" FROM "mesh_store_reader"`,
		`REVOKE ALL ON TABLE "public"."vault" FROM "mesh_store_reader"`,
		`RESET ROLE`,
		`SET ROLE "owner1"`,
		`REVOKE ALL ON TABLE "public"."vault" FROM "mesh_store_reader"`,
		`RESET ROLE`,
	}
	if strings.Join(stmts, "\n") != strings.Join(want, "\n") {
		t.Fatalf("got\n%s\nwant\n%s", strings.Join(stmts, "\n"), strings.Join(want, "\n"))
	}
	if left := unwanted(h, nil); strings.Join(left, ",") != "public.vault,public.vault.name" {
		t.Fatal(left)
	}
}

// The role is altered only when it drifted: the statement carries the password.
func TestTheReadersRoleIsAlteredOnlyWhenItDrifted(t *testing.T) {
	conn := admin
	conn.ReaderPassword = "r"
	for _, c := range []struct {
		row   []string
		alter bool
	}{
		{[]string{"f", "f", "f", "f", "f", "t", "t", "t", "default_transaction_read_only=on,statement_timeout=60s"}, false},
		{[]string{"f", "f", "f", "f", "t", "t", "t", "t", "default_transaction_read_only=on,statement_timeout=60s"}, true},
		{[]string{"f", "f", "f", "f", "f", "t", "t", "t", "statement_timeout=60s"}, true},
	} {
		f, cl := newFake(conn)
		f.answer(`^SELECT rolsuper`, []string{"a", "b", "c", "d", "e", "f", "g", "h", "i"}, c.row)
		if err := cl.ensureReaderRole(ctx); err != nil {
			t.Fatal(err)
		}
		altered := false
		for _, k := range f.calls {
			if strings.HasPrefix(k.SQL, "ALTER ROLE") && strings.Contains(k.SQL, "PASSWORD") {
				altered = true
			}
		}
		if altered != c.alter {
			t.Errorf("%v: altered %v", c.row, altered)
		}
	}
}

func TestASchemaTheReaderCannotUseIsGrantedUsage(t *testing.T) {
	decl, _ := ParseReadable(map[string]any{"readable": []any{"audit.events"}})
	stmts, _ := planDatabase(decl, []column{{"audit", "events", "at"}}, newHeld(),
		map[string]bool{"audit": true})
	if len(stmts) != 2 || stmts[0] != `GRANT USAGE ON SCHEMA "audit" TO "mesh_store_reader"` {
		t.Fatal(stmts)
	}
}

func TestAStatementNamingASecretIsRefusedBeforeItIsSent(t *testing.T) {
	conn := admin
	conn.ReaderPassword = "r"
	for _, c := range []struct{ db, sql string }{
		{"identity", "select private from signing_key"},
		{"identity", "SELECT * FROM Authority"},
		{"identity", `select * from "signing_key"`},
		{"inventory", "select count(*) from module_secret"},
		{"inventory", "select * from bus_membership"},
		{"inventory", "select query_to_xml('select * from secret', true, true, '')"},
		{"anything", "select operator_sealed from t"},
		{"anything", "select client_secret from t"},
		{"anything", "select * from pg_authid"},
	} {
		f, cl := newFake(conn)
		_, err := cl.ReadOnlyQuery(ctx, c.db, c.sql)
		if err == nil || !strings.Contains(err.Error(), "issue 337") {
			t.Errorf("%s in %s: %v", c.sql, c.db, err)
		}
		if len(f.calls) != 0 {
			t.Errorf("%s in %s was sent: %v", c.sql, c.db, f.calls)
		}
	}
	// Ordinary words pass: a table called secret elsewhere is the privileges' to refuse, not the words'.
	for _, ok := range []string{"select jsonb_object_keys(x) from t", "select name, created from node", "select token_count from usage"} {
		if why := refusedStatement("inventory", ok); why != "" {
			t.Errorf("%s: %s", ok, why)
		}
	}
}

func TestAQueryIsRefusedWhileTheReaderCanReadASecret(t *testing.T) {
	conn := admin
	conn.ReaderPassword = "r"
	f, c := newFake(conn)
	f.answer(`has_column_privilege`, []string{"nspname", "relname", "attname"},
		[]string{"public", "users", "name"}, []string{"public", "users", "password_hash"})
	f.answer(`^SELECT name FROM users`, []string{"name"}, []string{"a"})
	_, err := c.ReadOnlyQuery(ctx, "app", "SELECT name FROM users")
	if err == nil || !strings.Contains(err.Error(), "public.users.password_hash") || strings.Contains(err.Error(), "users.name") {
		t.Fatal(err)
	}
	for _, k := range f.calls {
		if k.SQL == "SELECT name FROM users" {
			t.Fatal("the statement ran")
		}
	}
	// The probe runs as the reader, in the database asked about.
	var probed bool
	for _, k := range f.calls {
		if strings.Contains(k.SQL, "has_column_privilege") && k.User == Reader && k.Database == "app" {
			probed = true
		}
	}
	if !probed {
		t.Fatal("not probed as the reader")
	}
}

func TestDeclarationsFromTheContributionsFile(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "mesh.json")
	doc := `{"requirement":"postgres-database","given":[
		{"as":"letta","secret":"s","values":{"name":"letta","readable":["agents"]}},
		{"as":"gitea","secret":"s","values":{"name":"gitea"}},
		{"as":"bad","secret":"s","values":{"readable":"agents"}}]}`
	if err := os.WriteFile(path, []byte(doc), 0o600); err != nil {
		t.Fatal(err)
	}
	d, err := DeclarationsFrom(path)
	if err != nil {
		t.Fatal(err)
	}
	if len(d.Tables) != 1 || d.Tables["letta"][0].Table != "agents" || d.Malformed["bad"] == "" {
		t.Fatalf("%+v", d)
	}
	if _, err := DeclarationsFrom(filepath.Join(dir, "absent")); err == nil {
		t.Fatal("an unreadable file is said")
	}
}

func TestReconcileRevokesTheRolesTheReaderHoldsAndGrantsPerDatabase(t *testing.T) {
	conn := admin
	conn.ReaderPassword = "r"
	f, c := newFake(conn)
	f.answer(`FROM pg_auth_members`, []string{"rolname"}, []string{"pg_read_all_data"})
	f.answer(`SELECT datname FROM pg_database WHERE datallowconn`, []string{"datname"}, []string{"identity"}, []string{"letta"})
	f.answer(`aclexplode\(c\.relacl\)`, []string{"nspname", "relname", "relkind"})
	f.answer(`aclexplode\(a\.attacl\)`, []string{"nspname", "relname", "attname", "privilege_type"})
	f.answer(`IN \(\('public', 'agents'\)\)`, []string{"nspname", "relname", "attname"},
		[]string{"public", "agents", "id"}, []string{"public", "agents", "api_key"})
	f.answer(`has_schema_privilege`, []string{"nspname"})
	rep := c.ReconcileReader(ctx, Declarations{Tables: map[string][]ReadableTable{"letta": {{Schema: "public", Table: "agents"}}}}, "")
	if rep.Problem != "" || len(rep.Databases) != 2 {
		t.Fatalf("%+v", rep)
	}
	var revoked, granted bool
	for _, k := range f.calls {
		if k.SQL == `REVOKE "pg_read_all_data" FROM "mesh_store_reader"` && k.Database == "postgres" {
			revoked = true
		}
		if k.SQL == `GRANT SELECT ("id") ON "public"."agents" TO "mesh_store_reader"` && k.Database == "letta" {
			granted = true
		}
		if strings.Contains(k.SQL, "GRANT") && k.Database == "identity" {
			t.Fatal("granted in an undeclared database:", k.SQL)
		}
		if strings.Contains(k.SQL, "GRANT") && strings.Contains(k.SQL, "api_key") {
			t.Fatal("granted a secret:", k.SQL)
		}
	}
	if !revoked || !granted {
		t.Fatalf("revoked %v, granted %v: %v", revoked, granted, f.calls)
	}
	if last := c.LastReaderReport(); len(last.Databases) != 2 || last.Databases[1].Refused == nil {
		t.Fatalf("%+v", last)
	}
}

// Every module in the catalogue that declares what the reader may read declares it in a shape the
// provider reads: a typo is found here, not as a refusal on a live store.
func TestTheCataloguesReadableDeclarationsParse(t *testing.T) {
	manifests, _ := filepath.Glob("../../../*/module.json")
	if len(manifests) == 0 {
		t.Skip("not in the catalogue")
	}
	for _, path := range manifests {
		raw, err := os.ReadFile(path)
		if err != nil {
			t.Fatal(err)
		}
		var m struct {
			Contributes map[string]json.RawMessage `json:"contributes"`
		}
		if json.Unmarshal(raw, &m) != nil {
			continue
		}
		var values map[string]any
		if json.Unmarshal(m.Contributes["postgres-database"], &values) != nil {
			continue
		}
		decl, err := ParseReadable(values)
		if err != nil {
			t.Errorf("%s: %v", path, err)
		}
		for _, d := range decl {
			if SecretName(d.Table) {
				t.Errorf("%s declares %s readable, whose name says it holds secrets; it would never be granted", path, d)
			}
		}
	}
}

// A change that fails reports nothing revoked, and says what is still held and why — every pass, not
// "revoked" every pass.
func TestAFailedChangeSaysWhatIsStillHeldAndRevokesNothing(t *testing.T) {
	conn := admin
	conn.ReaderPassword = "r"
	f, c := newFake(conn)
	f.answer(`^SELECT rolsuper`, []string{"a", "b", "c", "d", "e", "f", "g", "h", "i"},
		[]string{"f", "f", "f", "f", "f", "t", "t", "t", "default_transaction_read_only=on,statement_timeout=60s"})
	f.answer(`SELECT datname FROM pg_database WHERE datallowconn`, []string{"datname"}, []string{"app"})
	f.answer(`aclexplode\(c\.relacl\)`, []string{"nspname", "relname", "relkind", "privilege_type", "grantor"},
		[]string{"public", "vault", "r", "SELECT", "owner1"})
	f.answer(`aclexplode\(a\.attacl\)`, []string{"nspname", "relname", "attname", "privilege_type", "grantor"})
	f.fail(`^SET ROLE`, errors.New("server busy"))
	rep := c.ReconcileReader(ctx, Declarations{}, "")
	db := rep.Databases[0]
	if len(db.Revoked) != 0 {
		t.Fatalf("said revoked: %v", db.Revoked)
	}
	if !strings.Contains(db.Error, "STILL HELD") || !strings.Contains(db.Error, "public.vault") || !strings.Contains(db.Error, "server busy") {
		t.Fatal(db.Error)
	}
}