database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-07-18 17:42:25 +03:00
Code Activity

Add a security_barrier option for views.

Browse Source 
When a view is marked as a security barrier, it will not be pulled up
into the containing query, and no quals will be pushed down into it,
so that no function or operator chosen by the user can be applied to
rows not exposed by the view.  Views not configured with this
option cannot provide robust row-level security, but will perform far
better.

Patch by KaiGai Kohei; original problem report by Heikki Linnakangas
(in October 2009!).  Review (in earlier versions) by Noah Misch and
others.  Design advice by Tom Lane and myself.  Further review and
cleanup by me.
This commit is contained in:
Robert Haas
2011-12-22 16:15:57 -05:00
parent f90dd28062
commit 0e4611c023
22 changed files with 300 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/backend/access/common/reloptions.c
View File

@ -67,6 +67,14 @@ static relopt_bool boolRelOpts[] =
},
true
},
{
{
"security_barrier",
"View acts as a row security barrier",
RELOPT_KIND_VIEW
},
false
},
/* list terminator */
{{NULL}}
};
@ -781,6 +789,7 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc, Oid amoptions)
{
case RELKIND_RELATION:
case RELKIND_TOASTVALUE:
case RELKIND_VIEW:
case RELKIND_UNCATALOGED:
options = heap_reloptions(classForm->relkind, datum, false);
break;
@ -1139,7 +1148,9 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
{"autovacuum_vacuum_scale_factor", RELOPT_TYPE_REAL,
offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, vacuum_scale_factor)},
{"autovacuum_analyze_scale_factor", RELOPT_TYPE_REAL,
offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, analyze_scale_factor)}
offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, analyze_scale_factor)},
{"security_barrier", RELOPT_TYPE_BOOL,
offsetof(StdRdOptions, security_barrier)},
};
options = parseRelOptions(reloptions, validate, kind, &numoptions);
@ -1159,7 +1170,7 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
}
/*
* Parse options for heaps and toast tables.
* Parse options for heaps, views and toast tables.
*/
bytea *
heap_reloptions(char relkind, Datum reloptions, bool validate)
@ -1181,6 +1192,8 @@ heap_reloptions(char relkind, Datum reloptions, bool validate)
return (bytea *) rdopts;
case RELKIND_RELATION:
return default_reloptions(reloptions, validate, RELOPT_KIND_HEAP);
case RELKIND_VIEW:
return default_reloptions(reloptions, validate, RELOPT_KIND_VIEW);
default:
/* other relkinds are not supported */
return NULL;

66
src/backend/commands/tablecmds.c
View File

@ -366,7 +366,9 @@ static void ATExecDropCluster(Relation rel, LOCKMODE lockmode);
static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
char *tablespacename, LOCKMODE lockmode);
static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
static void ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode);
static void ATExecSetRelOptions(Relation rel, List *defList,
AlterTableType operation,
LOCKMODE lockmode);
static void ATExecEnableDisableTrigger(Relation rel, char *trigname,
char fires_when, bool skip_system, LOCKMODE lockmode);
static void ATExecEnableDisableRule(Relation rel, char *rulename,
@ -2866,6 +2868,7 @@ AlterTableGetLockLevel(List *cmds)
case AT_DropCluster:
case AT_SetRelOptions:
case AT_ResetRelOptions:
case AT_ReplaceRelOptions:
case AT_SetOptions:
case AT_ResetOptions:
case AT_SetStorage:
@ -3094,8 +3097,9 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
pass = AT_PASS_MISC; /* doesn't actually matter */
break;
case AT_SetRelOptions: /* SET (...) */
case AT_ResetRelOptions: /* RESET (...) */
ATSimplePermissions(rel, ATT_TABLE | ATT_INDEX);
case AT_ResetRelOptions: /* RESET (...) */
case AT_ReplaceRelOptions: /* reset them all, then set just these */
ATSimplePermissions(rel, ATT_TABLE | ATT_INDEX | ATT_VIEW);
/* This command never recurses */
/* No command-specific prep needed */
pass = AT_PASS_MISC;
@ -3338,12 +3342,10 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
*/
break;
case AT_SetRelOptions: /* SET (...) */
ATExecSetRelOptions(rel, (List *) cmd->def, false, lockmode);
break;
case AT_ResetRelOptions: /* RESET (...) */
ATExecSetRelOptions(rel, (List *) cmd->def, true, lockmode);
case AT_ReplaceRelOptions: /* replace entire option list */
ATExecSetRelOptions(rel, (List *) cmd->def, cmd->subtype, lockmode);
break;
case AT_EnableTrig: /* ENABLE TRIGGER name */
ATExecEnableDisableTrigger(rel, cmd->name,
TRIGGER_FIRES_ON_ORIGIN, false, lockmode);
@ -8271,10 +8273,11 @@ ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, char *tablespacename, L
}
/*
* ALTER TABLE/INDEX SET (...) or RESET (...)
* Set, reset, or replace reloptions.
*/
static void
ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode)
ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
LOCKMODE lockmode)
{
Oid relid;
Relation pgclass;
@ -8288,28 +8291,44 @@ ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode
bool repl_repl[Natts_pg_class];
static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
if (defList == NIL)
if (defList == NIL && operation != AT_ReplaceRelOptions)
return; /* nothing to do */
pgclass = heap_open(RelationRelationId, RowExclusiveLock);
/* Get the old reloptions */
/* Fetch heap tuple */
relid = RelationGetRelid(rel);
tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
if (!HeapTupleIsValid(tuple))
elog(ERROR, "cache lookup failed for relation %u", relid);
datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions, &isnull);
if (operation == AT_ReplaceRelOptions)
{
/*
* If we're supposed to replace the reloptions list, we just pretend
* there were none before.
*/
datum = (Datum) 0;
isnull = true;
}
else
{
/* Get the old reloptions */
datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions,
&isnull);
}
/* Generate new proposed reloptions (text array) */
newOptions = transformRelOptions(isnull ? (Datum) 0 : datum,
defList, NULL, validnsps, false, isReset);
defList, NULL, validnsps, false,
operation == AT_ResetRelOptions);
/* Validate */
switch (rel->rd_rel->relkind)
{
case RELKIND_RELATION:
case RELKIND_TOASTVALUE:
case RELKIND_VIEW:
(void) heap_reloptions(rel->rd_rel->relkind, newOptions, true);
break;
case RELKIND_INDEX:
@ -8357,15 +8376,30 @@ ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode
toastrel = heap_open(toastid, lockmode);
/* Get the old reloptions */
/* Fetch heap tuple */
tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(toastid));
if (!HeapTupleIsValid(tuple))
elog(ERROR, "cache lookup failed for relation %u", toastid);
datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions, &isnull);
if (operation == AT_ReplaceRelOptions)
{
/*
* If we're supposed to replace the reloptions list, we just
* pretend there were none before.
*/
datum = (Datum) 0;
isnull = true;
}
else
{
/* Get the old reloptions */
datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions,
&isnull);
}
newOptions = transformRelOptions(isnull ? (Datum) 0 : datum,
defList, "toast", validnsps, false, isReset);
defList, "toast", validnsps, false,
operation == AT_ResetRelOptions);
(void) heap_reloptions(RELKIND_TOASTVALUE, newOptions, true);

26
src/backend/commands/view.c
View File

@ -32,6 +32,7 @@
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/syscache.h"
static void checkViewTupleDesc(TupleDesc newdesc, TupleDesc olddesc);
@ -98,7 +99,7 @@ isViewOnTempTable_walker(Node *node, void *context)
*/
static Oid
DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
Oid namespaceId)
Oid namespaceId, List *options)
{
Oid viewOid;
CreateStmt *createStmt = makeNode(CreateStmt);
@ -166,6 +167,8 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
{
Relation rel;
TupleDesc descriptor;
List *atcmds = NIL;
AlterTableCmd *atcmd;
/*
* Yes. Get exclusive lock on the existing view ...
@ -203,6 +206,15 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
descriptor = BuildDescForRelation(attrList);
checkViewTupleDesc(descriptor, rel->rd_att);
/*
* The new options list replaces the existing options list, even
* if it's empty.
*/
atcmd = makeNode(AlterTableCmd);
atcmd->subtype = AT_ReplaceRelOptions;
atcmd->def = (Node *) options;
atcmds = lappend(atcmds, atcmd);
/*
* If new attributes have been added, we must add pg_attribute entries
* for them. It is convenient (although overkill) to use the ALTER
@ -210,14 +222,11 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
*/
if (list_length(attrList) > rel->rd_att->natts)
{
List *atcmds = NIL;
ListCell *c;
int skip = rel->rd_att->natts;
foreach(c, attrList)
{
AlterTableCmd *atcmd;
if (skip > 0)
{
skip--;
@ -228,9 +237,11 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
atcmd->def = (Node *) lfirst(c);
atcmds = lappend(atcmds, atcmd);
}
AlterTableInternal(viewOid, atcmds, true);
}
/* OK, let's do it. */
AlterTableInternal(viewOid, atcmds, true);
/*
* Seems okay, so return the OID of the pre-existing view.
*/
@ -250,7 +261,8 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
createStmt->tableElts = attrList;
createStmt->inhRelations = NIL;
createStmt->constraints = NIL;
createStmt->options = list_make1(defWithOids(false));
createStmt->options = options;
createStmt->options = lappend(options, defWithOids(false));
createStmt->oncommit = ONCOMMIT_NOOP;
createStmt->tablespacename = NULL;
createStmt->if_not_exists = false;
@ -513,7 +525,7 @@ DefineView(ViewStmt *stmt, const char *queryString)
* aborted.
*/
viewOid = DefineVirtualRelation(view, viewParse->targetList,
stmt->replace, namespaceId);
stmt->replace, namespaceId, stmt->options);
/*
* The relation we have just created is not visible to any other commands

1
src/backend/nodes/copyfuncs.c
View File

@ -1965,6 +1965,7 @@ _copyRangeTblEntry(const RangeTblEntry *from)
COPY_SCALAR_FIELD(relid);
COPY_SCALAR_FIELD(relkind);
COPY_NODE_FIELD(subquery);
COPY_SCALAR_FIELD(security_barrier);
COPY_SCALAR_FIELD(jointype);
COPY_NODE_FIELD(joinaliasvars);
COPY_NODE_FIELD(funcexpr);

1
src/backend/nodes/equalfuncs.c
View File

@ -2228,6 +2228,7 @@ _equalRangeTblEntry(const RangeTblEntry *a, const RangeTblEntry *b)
COMPARE_SCALAR_FIELD(relid);
COMPARE_SCALAR_FIELD(relkind);
COMPARE_NODE_FIELD(subquery);
COMPARE_SCALAR_FIELD(security_barrier);
COMPARE_SCALAR_FIELD(jointype);
COMPARE_NODE_FIELD(joinaliasvars);
COMPARE_NODE_FIELD(funcexpr);

1
src/backend/nodes/outfuncs.c
View File

@ -2321,6 +2321,7 @@ _outRangeTblEntry(StringInfo str, const RangeTblEntry *node)
break;
case RTE_SUBQUERY:
WRITE_NODE_FIELD(subquery);
WRITE_BOOL_FIELD(security_barrier);
break;
case RTE_JOIN:
WRITE_ENUM_FIELD(jointype, JoinType);

1
src/backend/nodes/readfuncs.c
View File

@ -1192,6 +1192,7 @@ _readRangeTblEntry(void)
break;
case RTE_SUBQUERY:
READ_NODE_FIELD(subquery);
READ_BOOL_FIELD(security_barrier);
break;
case RTE_JOIN:
READ_ENUM_FIELD(jointype, JoinType);

14
src/backend/optimizer/path/allpaths.c
View File

@ -744,6 +744,11 @@ set_subquery_pathlist(PlannerInfo *root, RelOptInfo *rel,
* pseudoconstant clauses; better to have the gating node above the
* subquery.
*
* Also, if the sub-query has "security_barrier" flag, it means the
* sub-query originated from a view that must enforce row-level security.
* We must not push down quals in order to avoid information leaks, either
* via side-effects or error output.
*
* Non-pushed-down clauses will get evaluated as qpquals of the
* SubqueryScan node.
*
@ -762,7 +767,16 @@ set_subquery_pathlist(PlannerInfo *root, RelOptInfo *rel,
RestrictInfo *rinfo = (RestrictInfo *) lfirst(l);
Node *clause = (Node *) rinfo->clause;
/*
* XXX. You might wonder why we're testing rte->security_barrier
* qual-by-qual here rather than hoisting the test up into the
* surrounding if statement; after all, the answer will be the
* same for all quals. The answer is that we expect to shortly
* change this logic to allow pushing down some quals that use only
* "leakproof" operators even through a security barrier.
*/
if (!rinfo->pseudoconstant &&
!rte->security_barrier &&
qual_is_pushdown_safe(subquery, rti, clause, differentTypes))
{
/* Push it down */

2
src/backend/optimizer/prep/prepjointree.c
View File

@ -543,6 +543,7 @@ pull_up_subqueries(PlannerInfo *root, Node *jtnode,
*/
if (rte->rtekind == RTE_SUBQUERY &&
is_simple_subquery(rte->subquery) &&
!rte->security_barrier &&
(containing_appendrel == NULL ||
is_safe_append_member(rte->subquery)))
return pull_up_simple_subquery(root, jtnode, rte,
@ -712,6 +713,7 @@ pull_up_simple_subquery(PlannerInfo *root, Node *jtnode, RangeTblEntry *rte,
* pull_up_subqueries.
*/
if (is_simple_subquery(subquery) &&
!rte->security_barrier &&
(containing_appendrel == NULL || is_safe_append_member(subquery)))
{
/* good to go */

10
src/backend/parser/gram.y
View File

@ -7322,26 +7322,28 @@ transaction_mode_list_or_empty:
*
*****************************************************************************/
ViewStmt: CREATE OptTemp VIEW qualified_name opt_column_list
ViewStmt: CREATE OptTemp VIEW qualified_name opt_column_list opt_reloptions
AS SelectStmt opt_check_option
{
ViewStmt *n = makeNode(ViewStmt);
n->view = $4;
n->view->relpersistence = $2;
n->aliases = $5;
n->query = $7;
n->query = $8;
n->replace = false;
n->options = $6;
$$ = (Node *) n;
}
| CREATE OR REPLACE OptTemp VIEW qualified_name opt_column_list
| CREATE OR REPLACE OptTemp VIEW qualified_name opt_column_list opt_reloptions
AS SelectStmt opt_check_option
{
ViewStmt *n = makeNode(ViewStmt);
n->view = $6;
n->view->relpersistence = $4;
n->aliases = $7;
n->query = $9;
n->query = $10;
n->replace = true;
n->options = $8;
$$ = (Node *) n;
}
;

1
src/backend/rewrite/rewriteHandler.c
View File

@ -1382,6 +1382,7 @@ ApplyRetrieveRule(Query *parsetree,
rte->rtekind = RTE_SUBQUERY;
rte->relid = InvalidOid;
rte->security_barrier = RelationIsSecurityView(relation);
rte->subquery = rule_action;
rte->inh = false; /* must not be set for a subquery */

13
src/backend/utils/adt/selfuncs.c
View File

@ -4372,6 +4372,19 @@ examine_simple_variable(PlannerInfo *root, Var *var,
subquery->distinctClause)
return;
/*
* If the sub-query originated from a view with the security_barrier
* attribute, we treat it as a black-box from outside of the view.
* This is probably a harsher restriction than necessary; it's
* certainly OK for the selectivity estimator (which is a C function,
* and therefore omnipotent anyway) to look at the statistics. But
* many selectivity estimators will happily *invoke the operator
* function* to try to work out a good estimate - and that's not OK.
* So for now, we do this.
*/
if (rte->security_barrier)
return;
/*
* OK, fetch RelOptInfo for subquery. Note that we don't change the
* rel returned in vardata, since caller expects it to be a rel of the

1
src/backend/utils/cache/relcache.c vendored
View File

@ -374,6 +374,7 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
case RELKIND_RELATION:
case RELKIND_TOASTVALUE:
case RELKIND_INDEX:
case RELKIND_VIEW:
break;
default:
return;