TLA Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * schemacmds.c
4 : * schema creation/manipulation commands
5 : *
6 : * Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
7 : * Portions Copyright (c) 1994, Regents of the University of California
8 : *
9 : *
10 : * IDENTIFICATION
11 : * src/backend/commands/schemacmds.c
12 : *
13 : *-------------------------------------------------------------------------
14 : */
15 : #include "postgres.h"
16 :
17 : #include "access/htup_details.h"
18 : #include "access/table.h"
19 : #include "access/xact.h"
20 : #include "catalog/catalog.h"
21 : #include "catalog/dependency.h"
22 : #include "catalog/indexing.h"
23 : #include "catalog/namespace.h"
24 : #include "catalog/objectaccess.h"
25 : #include "catalog/pg_authid.h"
26 : #include "catalog/pg_database.h"
27 : #include "catalog/pg_namespace.h"
28 : #include "commands/dbcommands.h"
29 : #include "commands/event_trigger.h"
30 : #include "commands/schemacmds.h"
31 : #include "miscadmin.h"
32 : #include "parser/parse_utilcmd.h"
33 : #include "tcop/utility.h"
34 : #include "utils/acl.h"
35 : #include "utils/builtins.h"
36 : #include "utils/rel.h"
37 : #include "utils/syscache.h"
38 :
39 : static void AlterSchemaOwner_internal(HeapTuple tup, Relation rel, Oid newOwnerId);
40 :
41 : /*
42 : * CREATE SCHEMA
43 : *
44 : * Note: caller should pass in location information for the whole
45 : * CREATE SCHEMA statement, which in turn we pass down as the location
46 : * of the component commands. This comports with our general plan of
47 : * reporting location/len for the whole command even when executing
48 : * a subquery.
49 : */
50 : Oid
51 GIC 674 : CreateSchemaCommand(CreateSchemaStmt *stmt, const char *queryString,
52 ECB : int stmt_location, int stmt_len)
53 : {
54 GIC 674 : const char *schemaName = stmt->schemaname;
55 ECB : Oid namespaceId;
56 : OverrideSearchPath *overridePath;
57 : List *parsetree_list;
58 : ListCell *parsetree_item;
59 : Oid owner_uid;
60 : Oid saved_uid;
61 : int save_sec_context;
62 : AclResult aclresult;
63 : ObjectAddress address;
64 :
65 GIC 674 : GetUserIdAndSecContext(&saved_uid, &save_sec_context);
66 ECB :
67 : /*
68 : * Who is supposed to own the new schema?
69 : */
70 GIC 674 : if (stmt->authrole)
71 CBC 34 : owner_uid = get_rolespec_oid(stmt->authrole, false);
72 ECB : else
73 GIC 640 : owner_uid = saved_uid;
74 ECB :
75 : /* fill schema name with the user name if not specified */
76 GIC 668 : if (!schemaName)
77 ECB : {
78 : HeapTuple tuple;
79 :
80 UIC 0 : tuple = SearchSysCache1(AUTHOID, ObjectIdGetDatum(owner_uid));
81 UBC 0 : if (!HeapTupleIsValid(tuple))
82 0 : elog(ERROR, "cache lookup failed for role %u", owner_uid);
83 EUB : schemaName =
84 UIC 0 : pstrdup(NameStr(((Form_pg_authid) GETSTRUCT(tuple))->rolname));
85 UBC 0 : ReleaseSysCache(tuple);
86 EUB : }
87 :
88 : /*
89 : * To create a schema, must have schema-create privilege on the current
90 : * database and must be able to become the target role (this does not
91 : * imply that the target role itself must have create-schema privilege).
92 : * The latter provision guards against "giveaway" attacks. Note that a
93 : * superuser will always have both of these privileges a fortiori.
94 : */
95 GNC 668 : aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, saved_uid, ACL_CREATE);
96 CBC 668 : if (aclresult != ACLCHECK_OK)
97 LBC 0 : aclcheck_error(aclresult, OBJECT_DATABASE,
98 UBC 0 : get_database_name(MyDatabaseId));
99 EUB :
100 GNC 668 : check_can_set_role(saved_uid, owner_uid);
101 ECB :
102 : /* Additional check to protect reserved schema names */
103 GIC 665 : if (!allowSystemTableMods && IsReservedName(schemaName))
104 CBC 1 : ereport(ERROR,
105 ECB : (errcode(ERRCODE_RESERVED_NAME),
106 : errmsg("unacceptable schema name \"%s\"", schemaName),
107 : errdetail("The prefix \"pg_\" is reserved for system schemas.")));
108 :
109 : /*
110 : * If if_not_exists was given and the schema already exists, bail out.
111 : * (Note: we needn't check this when not if_not_exists, because
112 : * NamespaceCreate will complain anyway.) We could do this before making
113 : * the permissions checks, but since CREATE TABLE IF NOT EXISTS makes its
114 : * creation-permission check first, we do likewise.
115 : */
116 GIC 664 : if (stmt->if_not_exists)
117 ECB : {
118 GIC 17 : namespaceId = get_namespace_oid(schemaName, true);
119 CBC 17 : if (OidIsValid(namespaceId))
120 ECB : {
121 : /*
122 : * If we are in an extension script, insist that the pre-existing
123 : * object be a member of the extension, to avoid security risks.
124 : */
125 GIC 12 : ObjectAddressSet(address, NamespaceRelationId, namespaceId);
126 CBC 12 : checkMembershipInCurrentExtension(&address);
127 ECB :
128 : /* OK to skip */
129 GIC 11 : ereport(NOTICE,
130 ECB : (errcode(ERRCODE_DUPLICATE_SCHEMA),
131 : errmsg("schema \"%s\" already exists, skipping",
132 : schemaName)));
133 GIC 11 : return InvalidOid;
134 ECB : }
135 : }
136 :
137 : /*
138 : * If the requested authorization is different from the current user,
139 : * temporarily set the current user so that the object(s) will be created
140 : * with the correct ownership.
141 : *
142 : * (The setting will be restored at the end of this routine, or in case of
143 : * error, transaction abort will clean things up.)
144 : */
145 GIC 652 : if (saved_uid != owner_uid)
146 CBC 17 : SetUserIdAndSecContext(owner_uid,
147 ECB : save_sec_context | SECURITY_LOCAL_USERID_CHANGE);
148 :
149 : /* Create the schema's namespace */
150 GIC 652 : namespaceId = NamespaceCreate(schemaName, owner_uid, false);
151 ECB :
152 : /* Advance cmd counter to make the namespace visible */
153 GIC 649 : CommandCounterIncrement();
154 ECB :
155 : /*
156 : * Temporarily make the new namespace be the front of the search path, as
157 : * well as the default creation target namespace. This will be undone at
158 : * the end of this routine, or upon error.
159 : */
160 GIC 649 : overridePath = GetOverrideSearchPath(CurrentMemoryContext);
161 CBC 649 : overridePath->schemas = lcons_oid(namespaceId, overridePath->schemas);
162 ECB : /* XXX should we clear overridePath->useTemp? */
163 GIC 649 : PushOverrideSearchPath(overridePath);
164 ECB :
165 : /*
166 : * Report the new schema to possibly interested event triggers. Note we
167 : * must do this here and not in ProcessUtilitySlow because otherwise the
168 : * objects created below are reported before the schema, which would be
169 : * wrong.
170 : */
171 GIC 649 : ObjectAddressSet(address, NamespaceRelationId, namespaceId);
172 CBC 649 : EventTriggerCollectSimpleCommand(address, InvalidObjectAddress,
173 ECB : (Node *) stmt);
174 :
175 : /*
176 : * Examine the list of commands embedded in the CREATE SCHEMA command, and
177 : * reorganize them into a sequentially executable order with no forward
178 : * references. Note that the result is still a list of raw parsetrees ---
179 : * we cannot, in general, run parse analysis on one statement until we
180 : * have actually executed the prior ones.
181 : */
182 GIC 649 : parsetree_list = transformCreateSchemaStmt(stmt);
183 ECB :
184 : /*
185 : * Execute each command contained in the CREATE SCHEMA. Since the grammar
186 : * allows only utility commands in CREATE SCHEMA, there is no need to pass
187 : * them through parse_analyze_*() or the rewriter; we can just hand them
188 : * straight to ProcessUtility.
189 : */
190 GIC 821 : foreach(parsetree_item, parsetree_list)
191 ECB : {
192 GIC 175 : Node *stmt = (Node *) lfirst(parsetree_item);
193 ECB : PlannedStmt *wrapper;
194 :
195 : /* need to make a wrapper PlannedStmt */
196 GIC 175 : wrapper = makeNode(PlannedStmt);
197 CBC 175 : wrapper->commandType = CMD_UTILITY;
198 175 : wrapper->canSetTag = false;
199 175 : wrapper->utilityStmt = stmt;
200 175 : wrapper->stmt_location = stmt_location;
201 175 : wrapper->stmt_len = stmt_len;
202 ECB :
203 : /* do this step */
204 GIC 175 : ProcessUtility(wrapper,
205 ECB : queryString,
206 : false,
207 : PROCESS_UTILITY_SUBCOMMAND,
208 : NULL,
209 : NULL,
210 : None_Receiver,
211 : NULL);
212 :
213 : /* make sure later steps can see the object created here */
214 GIC 172 : CommandCounterIncrement();
215 ECB : }
216 :
217 : /* Reset search path to normal state */
218 GIC 646 : PopOverrideSearchPath();
219 ECB :
220 : /* Reset current user and security context */
221 GIC 646 : SetUserIdAndSecContext(saved_uid, save_sec_context);
222 ECB :
223 GIC 646 : return namespaceId;
224 ECB : }
225 :
226 :
227 : /*
228 : * Rename schema
229 : */
230 : ObjectAddress
231 GIC 10 : RenameSchema(const char *oldname, const char *newname)
232 ECB : {
233 : Oid nspOid;
234 : HeapTuple tup;
235 : Relation rel;
236 : AclResult aclresult;
237 : ObjectAddress address;
238 : Form_pg_namespace nspform;
239 :
240 GIC 10 : rel = table_open(NamespaceRelationId, RowExclusiveLock);
241 ECB :
242 GIC 10 : tup = SearchSysCacheCopy1(NAMESPACENAME, CStringGetDatum(oldname));
243 CBC 10 : if (!HeapTupleIsValid(tup))
244 LBC 0 : ereport(ERROR,
245 EUB : (errcode(ERRCODE_UNDEFINED_SCHEMA),
246 : errmsg("schema \"%s\" does not exist", oldname)));
247 :
248 GIC 10 : nspform = (Form_pg_namespace) GETSTRUCT(tup);
249 CBC 10 : nspOid = nspform->oid;
250 ECB :
251 : /* make sure the new name doesn't exist */
252 GIC 10 : if (OidIsValid(get_namespace_oid(newname, true)))
253 LBC 0 : ereport(ERROR,
254 EUB : (errcode(ERRCODE_DUPLICATE_SCHEMA),
255 : errmsg("schema \"%s\" already exists", newname)));
256 :
257 : /* must be owner */
258 GNC 10 : if (!object_ownercheck(NamespaceRelationId, nspOid, GetUserId()))
259 LBC 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
260 EUB : oldname);
261 :
262 : /* must have CREATE privilege on database */
263 GNC 10 : aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, GetUserId(), ACL_CREATE);
264 CBC 10 : if (aclresult != ACLCHECK_OK)
265 LBC 0 : aclcheck_error(aclresult, OBJECT_DATABASE,
266 UBC 0 : get_database_name(MyDatabaseId));
267 EUB :
268 GIC 10 : if (!allowSystemTableMods && IsReservedName(newname))
269 LBC 0 : ereport(ERROR,
270 EUB : (errcode(ERRCODE_RESERVED_NAME),
271 : errmsg("unacceptable schema name \"%s\"", newname),
272 : errdetail("The prefix \"pg_\" is reserved for system schemas.")));
273 :
274 : /* rename */
275 GIC 10 : namestrcpy(&nspform->nspname, newname);
276 CBC 10 : CatalogTupleUpdate(rel, &tup->t_self, tup);
277 ECB :
278 GIC 10 : InvokeObjectPostAlterHook(NamespaceRelationId, nspOid, 0);
279 ECB :
280 GIC 10 : ObjectAddressSet(address, NamespaceRelationId, nspOid);
281 ECB :
282 GIC 10 : table_close(rel, NoLock);
283 CBC 10 : heap_freetuple(tup);
284 ECB :
285 GIC 10 : return address;
286 ECB : }
287 :
288 : void
289 GNC 3 : AlterSchemaOwner_oid(Oid schemaoid, Oid newOwnerId)
290 ECB : {
291 : HeapTuple tup;
292 : Relation rel;
293 :
294 GIC 3 : rel = table_open(NamespaceRelationId, RowExclusiveLock);
295 ECB :
296 GNC 3 : tup = SearchSysCache1(NAMESPACEOID, ObjectIdGetDatum(schemaoid));
297 CBC 3 : if (!HeapTupleIsValid(tup))
298 UNC 0 : elog(ERROR, "cache lookup failed for schema %u", schemaoid);
299 EUB :
300 GIC 3 : AlterSchemaOwner_internal(tup, rel, newOwnerId);
301 ECB :
302 GIC 3 : ReleaseSysCache(tup);
303 ECB :
304 GIC 3 : table_close(rel, RowExclusiveLock);
305 CBC 3 : }
306 ECB :
307 :
308 : /*
309 : * Change schema owner
310 : */
311 : ObjectAddress
312 GIC 26 : AlterSchemaOwner(const char *name, Oid newOwnerId)
313 ECB : {
314 : Oid nspOid;
315 : HeapTuple tup;
316 : Relation rel;
317 : ObjectAddress address;
318 : Form_pg_namespace nspform;
319 :
320 GIC 26 : rel = table_open(NamespaceRelationId, RowExclusiveLock);
321 ECB :
322 GIC 26 : tup = SearchSysCache1(NAMESPACENAME, CStringGetDatum(name));
323 CBC 26 : if (!HeapTupleIsValid(tup))
324 LBC 0 : ereport(ERROR,
325 EUB : (errcode(ERRCODE_UNDEFINED_SCHEMA),
326 : errmsg("schema \"%s\" does not exist", name)));
327 :
328 GIC 26 : nspform = (Form_pg_namespace) GETSTRUCT(tup);
329 CBC 26 : nspOid = nspform->oid;
330 ECB :
331 GIC 26 : AlterSchemaOwner_internal(tup, rel, newOwnerId);
332 ECB :
333 GIC 26 : ObjectAddressSet(address, NamespaceRelationId, nspOid);
334 ECB :
335 GIC 26 : ReleaseSysCache(tup);
336 ECB :
337 GIC 26 : table_close(rel, RowExclusiveLock);
338 ECB :
339 GIC 26 : return address;
340 ECB : }
341 :
342 : static void
343 GIC 29 : AlterSchemaOwner_internal(HeapTuple tup, Relation rel, Oid newOwnerId)
344 ECB : {
345 : Form_pg_namespace nspForm;
346 :
347 GIC 29 : Assert(tup->t_tableOid == NamespaceRelationId);
348 CBC 29 : Assert(RelationGetRelid(rel) == NamespaceRelationId);
349 ECB :
350 GIC 29 : nspForm = (Form_pg_namespace) GETSTRUCT(tup);
351 ECB :
352 : /*
353 : * If the new owner is the same as the existing owner, consider the
354 : * command to have succeeded. This is for dump restoration purposes.
355 : */
356 GIC 29 : if (nspForm->nspowner != newOwnerId)
357 ECB : {
358 : Datum repl_val[Natts_pg_namespace];
359 : bool repl_null[Natts_pg_namespace];
360 : bool repl_repl[Natts_pg_namespace];
361 : Acl *newAcl;
362 : Datum aclDatum;
363 : bool isNull;
364 : HeapTuple newtuple;
365 : AclResult aclresult;
366 :
367 : /* Otherwise, must be owner of the existing object */
368 GNC 20 : if (!object_ownercheck(NamespaceRelationId, nspForm->oid, GetUserId()))
369 LBC 0 : aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SCHEMA,
370 UBC 0 : NameStr(nspForm->nspname));
371 EUB :
372 : /* Must be able to become new owner */
373 GNC 20 : check_can_set_role(GetUserId(), newOwnerId);
374 ECB :
375 : /*
376 : * must have create-schema rights
377 : *
378 : * NOTE: This is different from other alter-owner checks in that the
379 : * current user is checked for create privileges instead of the
380 : * destination owner. This is consistent with the CREATE case for
381 : * schemas. Because superusers will always have this right, we need
382 : * no special case for them.
383 : */
384 GNC 20 : aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId, GetUserId(),
385 ECB : ACL_CREATE);
386 GIC 20 : if (aclresult != ACLCHECK_OK)
387 LBC 0 : aclcheck_error(aclresult, OBJECT_DATABASE,
388 UBC 0 : get_database_name(MyDatabaseId));
389 EUB :
390 GIC 20 : memset(repl_null, false, sizeof(repl_null));
391 CBC 20 : memset(repl_repl, false, sizeof(repl_repl));
392 ECB :
393 GIC 20 : repl_repl[Anum_pg_namespace_nspowner - 1] = true;
394 CBC 20 : repl_val[Anum_pg_namespace_nspowner - 1] = ObjectIdGetDatum(newOwnerId);
395 ECB :
396 : /*
397 : * Determine the modified ACL for the new owner. This is only
398 : * necessary when the ACL is non-null.
399 : */
400 GIC 20 : aclDatum = SysCacheGetAttr(NAMESPACENAME, tup,
401 ECB : Anum_pg_namespace_nspacl,
402 : &isNull);
403 GIC 20 : if (!isNull)
404 ECB : {
405 GIC 2 : newAcl = aclnewowner(DatumGetAclP(aclDatum),
406 ECB : nspForm->nspowner, newOwnerId);
407 GIC 2 : repl_repl[Anum_pg_namespace_nspacl - 1] = true;
408 CBC 2 : repl_val[Anum_pg_namespace_nspacl - 1] = PointerGetDatum(newAcl);
409 ECB : }
410 :
411 GIC 20 : newtuple = heap_modify_tuple(tup, RelationGetDescr(rel), repl_val, repl_null, repl_repl);
412 ECB :
413 GIC 20 : CatalogTupleUpdate(rel, &newtuple->t_self, newtuple);
414 ECB :
415 GIC 20 : heap_freetuple(newtuple);
416 ECB :
417 : /* Update owner dependency reference */
418 GIC 20 : changeDependencyOnOwner(NamespaceRelationId, nspForm->oid,
419 ECB : newOwnerId);
420 : }
421 :
422 GIC 29 : InvokeObjectPostAlterHook(NamespaceRelationId,
423 ECB : nspForm->oid, 0);
424 GIC 29 : }
|
