granicus.if.org Git - postgresql/blob - src/backend/catalog/aclchk.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * aclchk.c
   4  *        Routines to check access control permissions.
   5  *
   6  * Portions Copyright (c) 1996-2001, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $Header: /cvsroot/pgsql/src/backend/catalog/aclchk.c,v 1.53 2001/11/05 17:46:24 momjian Exp $
  12  *
  13  * NOTES
  14  *        See acl.h.
  15  *
  16  *-------------------------------------------------------------------------
  17  */
  18 #include "postgres.h"
  19
  20 #include "access/heapam.h"
  21 #include "catalog/catalog.h"
  22 #include "catalog/catname.h"
  23 #include "catalog/indexing.h"
  24 #include "catalog/pg_aggregate.h"
  25 #include "catalog/pg_group.h"
  26 #include "catalog/pg_operator.h"
  27 #include "catalog/pg_proc.h"
  28 #include "catalog/pg_shadow.h"
  29 #include "catalog/pg_type.h"
  30 #include "miscadmin.h"
  31 #include "parser/parse_agg.h"
  32 #include "parser/parse_func.h"
  33 #include "utils/acl.h"
  34 #include "utils/syscache.h"
  35 #include "utils/temprel.h"
  36
  37 static int32 aclcheck(Acl *acl, AclId id, AclIdType idtype, AclMode mode);
  38
  39 /* warning messages, now more explicit. */
  40 /* MUST correspond to the order of the ACLCHK_* result codes in acl.h. */
  41 char       *aclcheck_error_strings[] = {
  42         "No error.",
  43         "Permission denied.",
  44         "Table does not exist.",
  45         "Must be table owner."
  46 };
  47
  48
  49 #ifdef ACLDEBUG
  50 static
  51 dumpacl(Acl *acl)
  52 {
  53         int                     i;
  54         AclItem    *aip;
  55
  56         elog(DEBUG, "acl size = %d, # acls = %d",
  57                  ACL_SIZE(acl), ACL_NUM(acl));
  58         aip = ACL_DAT(acl);
  59         for (i = 0; i < ACL_NUM(acl); ++i)
  60                 elog(DEBUG, "   acl[%d]: %s", i,
  61                          DatumGetCString(DirectFunctionCall1(aclitemout,
  62                                                                                          PointerGetDatum(aip + i))));
  63 }
  64 #endif   /* ACLDEBUG */
  65
  66
  67 /*
  68  * Called to execute the utility commands GRANT and REVOKE
  69  */
  70 void
  71 ExecuteGrantStmt(GrantStmt *stmt)
  72 {
  73         List       *i;
  74         List       *j;
  75
  76         /* see comment in pg_type.h */
  77         Assert(ACLITEMSIZE == sizeof(AclItem));
  78
  79         foreach(i, stmt->relnames)
  80         {
  81                 char       *relname = strVal(lfirst(i));
  82                 Relation        relation;
  83                 HeapTuple       tuple;
  84                 Form_pg_class pg_class_tuple;
  85                 Datum           aclDatum;
  86                 bool            isNull;
  87                 Acl                *old_acl;
  88                 Acl                *new_acl;
  89                 unsigned        i;
  90                 HeapTuple       newtuple;
  91                 Datum           values[Natts_pg_class];
  92                 char            nulls[Natts_pg_class];
  93                 char            replaces[Natts_pg_class];
  94
  95
  96                 if (!pg_ownercheck(GetUserId(), relname, RELNAME))
  97                         elog(ERROR, "permission denied");
  98
  99                 /* open pg_class */
 100                 relation = heap_openr(RelationRelationName, RowExclusiveLock);
 101                 tuple = SearchSysCache(RELNAME,
 102                                                            PointerGetDatum(relname),
 103                                                            0, 0, 0);
 104                 if (!HeapTupleIsValid(tuple))
 105                 {
 106                         heap_close(relation, RowExclusiveLock);
 107                         elog(ERROR, "relation \"%s\" not found",
 108                                  relname);
 109                 }
 110                 pg_class_tuple = (Form_pg_class) GETSTRUCT(tuple);
 111
 112                 if (pg_class_tuple->relkind == RELKIND_INDEX)
 113                         elog(ERROR, "\"%s\" is an index",
 114                                  relname);
 115
 116                 /*
 117                  * If there's no ACL, create a default using the pg_class.relowner
 118                  * field.
 119                  */
 120                 aclDatum = SysCacheGetAttr(RELNAME, tuple, Anum_pg_class_relacl,
 121                                                                    &isNull);
 122                 if (isNull)
 123                         old_acl = acldefault(relname, pg_class_tuple->relowner);
 124                 else
 125                         /* get a detoasted copy of the rel's ACL */
 126                         old_acl = DatumGetAclPCopy(aclDatum);
 127
 128 #ifdef ACLDEBUG
 129                 dumpacl(old_acl);
 130 #endif
 131                 new_acl = old_acl;
 132
 133                 foreach(j, stmt->grantees)
 134                 {
 135                         PrivGrantee *grantee = (PrivGrantee *) lfirst(j);
 136                         char       *granteeString;
 137                         char       *aclString;
 138                         AclItem aclitem;
 139                         unsigned        modechg;
 140
 141                         if (grantee->username)
 142                                 granteeString = aclmakeuser("U", grantee->username);
 143                         else if (grantee->groupname)
 144                                 granteeString = aclmakeuser("G", grantee->groupname);
 145                         else
 146                                 granteeString = aclmakeuser("A", "");
 147
 148                         aclString = makeAclString(stmt->privileges, granteeString,
 149                                                                           stmt->is_grant ? '+' : '-');
 150
 151                         /* Convert string ACL spec into internal form */
 152                         aclparse(aclString, &aclitem, &modechg);
 153                         new_acl = aclinsert3(new_acl, &aclitem, modechg);
 154 #ifdef ACLDEBUG
 155                         dumpacl(new_acl);
 156 #endif
 157                 }
 158
 159                 /* finished building new ACL value, now insert it */
 160                 for (i = 0; i < Natts_pg_class; ++i)
 161                 {
 162                         replaces[i] = ' ';
 163                         nulls[i] = ' ';         /* ignored if replaces[i]==' ' anyway */
 164                         values[i] = (Datum) NULL;       /* ignored if replaces[i]==' '
 165                                                                                  * anyway */
 166                 }
 167                 replaces[Anum_pg_class_relacl - 1] = 'r';
 168                 values[Anum_pg_class_relacl - 1] = PointerGetDatum(new_acl);
 169                 newtuple = heap_modifytuple(tuple, relation, values, nulls, replaces);
 170
 171                 ReleaseSysCache(tuple);
 172
 173                 simple_heap_update(relation, &newtuple->t_self, newtuple);
 174
 175                 {
 176                         /* keep the catalog indexes up to date */
 177                         Relation        idescs[Num_pg_class_indices];
 178
 179                         CatalogOpenIndices(Num_pg_class_indices, Name_pg_class_indices,
 180                                                            idescs);
 181                         CatalogIndexInsert(idescs, Num_pg_class_indices, relation, newtuple);
 182                         CatalogCloseIndices(Num_pg_class_indices, idescs);
 183                 }
 184
 185                 pfree(old_acl);
 186                 pfree(new_acl);
 187
 188                 heap_close(relation, RowExclusiveLock);
 189         }
 190 }
 191
 192
 193
 194 AclId
 195 get_grosysid(char *groname)
 196 {
 197         HeapTuple       tuple;
 198         AclId           id = 0;
 199
 200         tuple = SearchSysCache(GRONAME,
 201                                                    PointerGetDatum(groname),
 202                                                    0, 0, 0);
 203         if (HeapTupleIsValid(tuple))
 204         {
 205                 id = ((Form_pg_group) GETSTRUCT(tuple))->grosysid;
 206                 ReleaseSysCache(tuple);
 207         }
 208         else
 209                 elog(ERROR, "non-existent group \"%s\"", groname);
 210         return id;
 211 }
 212
 213 /*
 214  * Convert group ID to name, or return NULL if group can't be found
 215  */
 216 char *
 217 get_groname(AclId grosysid)
 218 {
 219         HeapTuple       tuple;
 220         char       *name = NULL;
 221
 222         tuple = SearchSysCache(GROSYSID,
 223                                                    ObjectIdGetDatum(grosysid),
 224                                                    0, 0, 0);
 225         if (HeapTupleIsValid(tuple))
 226         {
 227                 name = pstrdup(NameStr(((Form_pg_group) GETSTRUCT(tuple))->groname));
 228                 ReleaseSysCache(tuple);
 229         }
 230         return name;
 231 }
 232
 233 /*
 234  * Is user a member of group?
 235  */
 236 static bool
 237 in_group(AclId uid, AclId gid)
 238 {
 239         bool            result = false;
 240         HeapTuple       tuple;
 241         Datum           att;
 242         bool            isNull;
 243         IdList     *glist;
 244         AclId      *aidp;
 245         int                     i,
 246                                 num;
 247
 248         tuple = SearchSysCache(GROSYSID,
 249                                                    ObjectIdGetDatum(gid),
 250                                                    0, 0, 0);
 251         if (HeapTupleIsValid(tuple))
 252         {
 253                 att = SysCacheGetAttr(GROSYSID,
 254                                                           tuple,
 255                                                           Anum_pg_group_grolist,
 256                                                           &isNull);
 257                 if (!isNull)
 258                 {
 259                         /* be sure the IdList is not toasted */
 260                         glist = DatumGetIdListP(att);
 261                         /* scan it */
 262                         num = IDLIST_NUM(glist);
 263                         aidp = IDLIST_DAT(glist);
 264                         for (i = 0; i < num; ++i)
 265                         {
 266                                 if (aidp[i] == uid)
 267                                 {
 268                                         result = true;
 269                                         break;
 270                                 }
 271                         }
 272                         /* if IdList was toasted, free detoasted copy */
 273                         if ((Pointer) glist != DatumGetPointer(att))
 274                                 pfree(glist);
 275                 }
 276                 ReleaseSysCache(tuple);
 277         }
 278         else
 279                 elog(NOTICE, "in_group: group %u not found", gid);
 280         return result;
 281 }
 282
 283 /*
 284  * aclcheck
 285  *
 286  * Returns ACLCHECK_OK if the 'id' of type 'idtype' has ACL entries in 'acl'
 287  * to satisfy any one of the requirements of 'mode'.  Returns an appropriate
 288  * ACLCHECK_* error code otherwise.
 289  *
 290  * The ACL list is expected to be sorted in standard order.
 291  */
 292 static int32
 293 aclcheck(Acl *acl, AclId id, AclIdType idtype, AclMode mode)
 294 {
 295         AclItem    *aip,
 296                            *aidat;
 297         int                     i,
 298                                 num;
 299
 300         /*
 301          * If ACL is null, default to "OK" --- this should not happen, since
 302          * caller should have inserted appropriate default
 303          */
 304         if (!acl)
 305         {
 306                 elog(DEBUG, "aclcheck: null ACL, returning OK");
 307                 return ACLCHECK_OK;
 308         }
 309
 310         num = ACL_NUM(acl);
 311         aidat = ACL_DAT(acl);
 312
 313         /*
 314          * We'll treat the empty ACL like that, too, although this is more
 315          * like an error (i.e., you manually blew away your ACL array) -- the
 316          * system never creates an empty ACL, since there must always be a
 317          * "world" entry in the first slot.
 318          */
 319         if (num < 1)
 320         {
 321                 elog(DEBUG, "aclcheck: zero-length ACL, returning OK");
 322                 return ACLCHECK_OK;
 323         }
 324
 325         /*
 326          * "World" rights are applicable regardless of the passed-in ID, and
 327          * since they're much the cheapest to check, check 'em first.
 328          */
 329         if (aidat->ai_idtype != ACL_IDTYPE_WORLD)
 330                 elog(ERROR, "aclcheck: first entry in ACL is not 'world' entry");
 331         if (aidat->ai_mode & mode)
 332         {
 333 #ifdef ACLDEBUG
 334                 elog(DEBUG, "aclcheck: using world=%d", aidat->ai_mode);
 335 #endif
 336                 return ACLCHECK_OK;
 337         }
 338
 339         switch (idtype)
 340         {
 341                 case ACL_IDTYPE_UID:
 342                         /* See if permission is granted directly to user */
 343                         for (i = 1, aip = aidat + 1;            /* skip world entry */
 344                                  i < num && aip->ai_idtype == ACL_IDTYPE_UID;
 345                                  ++i, ++aip)
 346                         {
 347                                 if (aip->ai_id == id)
 348                                 {
 349 #ifdef ACLDEBUG
 350                                         elog(DEBUG, "aclcheck: found user %u/%d",
 351                                                  aip->ai_id, aip->ai_mode);
 352 #endif
 353                                         if (aip->ai_mode & mode)
 354                                                 return ACLCHECK_OK;
 355                                 }
 356                         }
 357                         /* See if he has the permission via any group */
 358                         for (;
 359                                  i < num && aip->ai_idtype == ACL_IDTYPE_GID;
 360                                  ++i, ++aip)
 361                         {
 362                                 if (aip->ai_mode & mode)
 363                                 {
 364                                         if (in_group(id, aip->ai_id))
 365                                         {
 366 #ifdef ACLDEBUG
 367                                                 elog(DEBUG, "aclcheck: found group %u/%d",
 368                                                          aip->ai_id, aip->ai_mode);
 369 #endif
 370                                                 return ACLCHECK_OK;
 371                                         }
 372                                 }
 373                         }
 374                         break;
 375                 case ACL_IDTYPE_GID:
 376                         /* Look for this group ID */
 377                         for (i = 1, aip = aidat + 1;            /* skip world entry */
 378                                  i < num && aip->ai_idtype == ACL_IDTYPE_UID;
 379                                  ++i, ++aip)
 380                                  /* skip UID entry */ ;
 381                         for (;
 382                                  i < num && aip->ai_idtype == ACL_IDTYPE_GID;
 383                                  ++i, ++aip)
 384                         {
 385                                 if (aip->ai_id == id)
 386                                 {
 387 #ifdef ACLDEBUG
 388                                         elog(DEBUG, "aclcheck: found group %u/%d",
 389                                                  aip->ai_id, aip->ai_mode);
 390 #endif
 391                                         if (aip->ai_mode & mode)
 392                                                 return ACLCHECK_OK;
 393                                 }
 394                         }
 395                         break;
 396                 case ACL_IDTYPE_WORLD:
 397                         /* Only check the world entry */
 398                         break;
 399                 default:
 400                         elog(ERROR, "aclcheck: bogus ACL id type: %d", idtype);
 401                         break;
 402         }
 403
 404         /* If get here, he doesn't have the privilege nohow */
 405         return ACLCHECK_NO_PRIV;
 406 }
 407
 408 /*
 409  * Exported routine for checking a user's access privileges to a table
 410  *
 411  * Returns an ACLCHECK_* result code.
 412  */
 413 int32
 414 pg_aclcheck(char *relname, Oid userid, AclMode mode)
 415 {
 416         int32           result;
 417         HeapTuple       tuple;
 418         char       *usename;
 419         Datum           aclDatum;
 420         bool            isNull;
 421         Acl                *acl;
 422
 423         /*
 424          * Validate userid, find out if he is superuser
 425          */
 426         tuple = SearchSysCache(SHADOWSYSID,
 427                                                    ObjectIdGetDatum(userid),
 428                                                    0, 0, 0);
 429         if (!HeapTupleIsValid(tuple))
 430                 elog(ERROR, "pg_aclcheck: invalid user id %u",
 431                          (unsigned) userid);
 432
 433         usename = NameStr(((Form_pg_shadow) GETSTRUCT(tuple))->usename);
 434
 435         /*
 436          * Deny anyone permission to update a system catalog unless
 437          * pg_shadow.usecatupd is set.  (This is to let superusers protect
 438          * themselves from themselves.)
 439          */
 440         if ((mode & (ACL_INSERT | ACL_UPDATE | ACL_DELETE)) &&
 441                 !allowSystemTableMods && IsSystemRelationName(relname) &&
 442                 !is_temp_relname(relname) &&
 443                 !((Form_pg_shadow) GETSTRUCT(tuple))->usecatupd)
 444         {
 445 #ifdef ACLDEBUG
 446                 elog(DEBUG, "pg_aclcheck: catalog update to \"%s\": permission denied",
 447                          relname);
 448 #endif
 449                 ReleaseSysCache(tuple);
 450                 return ACLCHECK_NO_PRIV;
 451         }
 452
 453         /*
 454          * Otherwise, superusers bypass all permission-checking.
 455          */
 456         if (((Form_pg_shadow) GETSTRUCT(tuple))->usesuper)
 457         {
 458 #ifdef ACLDEBUG
 459                 elog(DEBUG, "pg_aclcheck: \"%s\" is superuser",
 460                          usename);
 461 #endif
 462                 ReleaseSysCache(tuple);
 463                 return ACLCHECK_OK;
 464         }
 465
 466         ReleaseSysCache(tuple);
 467         /* caution: usename is inaccessible beyond this point... */
 468
 469         /*
 470          * Normal case: get the relation's ACL from pg_class
 471          */
 472         tuple = SearchSysCache(RELNAME,
 473                                                    PointerGetDatum(relname),
 474                                                    0, 0, 0);
 475         if (!HeapTupleIsValid(tuple))
 476                 elog(ERROR, "pg_aclcheck: class \"%s\" not found", relname);
 477
 478         aclDatum = SysCacheGetAttr(RELNAME, tuple, Anum_pg_class_relacl,
 479                                                            &isNull);
 480         if (isNull)
 481         {
 482                 /* No ACL, so build default ACL for rel */
 483                 AclId           ownerId;
 484
 485                 ownerId = ((Form_pg_class) GETSTRUCT(tuple))->relowner;
 486                 acl = acldefault(relname, ownerId);
 487                 aclDatum = (Datum) 0;
 488         }
 489         else
 490         {
 491                 /* detoast rel's ACL if necessary */
 492                 acl = DatumGetAclP(aclDatum);
 493         }
 494
 495         result = aclcheck(acl, userid, (AclIdType) ACL_IDTYPE_UID, mode);
 496
 497         /* if we have a detoasted copy, free it */
 498         if (acl && (Pointer) acl != DatumGetPointer(aclDatum))
 499                 pfree(acl);
 500
 501         ReleaseSysCache(tuple);
 502
 503         return result;
 504 }
 505
 506 /*
 507  * Check ownership of an object identified by name (which will be looked
 508  * up in the system cache identified by cacheid).
 509  *
 510  * Returns true if userid owns the item, or should be allowed to modify
 511  * the item as if he owned it.
 512  */
 513 bool
 514 pg_ownercheck(Oid userid,
 515                           const char *name,
 516                           int cacheid)
 517 {
 518         HeapTuple       tuple;
 519         AclId           owner_id;
 520         char       *usename;
 521
 522         tuple = SearchSysCache(SHADOWSYSID,
 523                                                    ObjectIdGetDatum(userid),
 524                                                    0, 0, 0);
 525         if (!HeapTupleIsValid(tuple))
 526                 elog(ERROR, "pg_ownercheck: invalid user id %u",
 527                          (unsigned) userid);
 528         usename = NameStr(((Form_pg_shadow) GETSTRUCT(tuple))->usename);
 529
 530         /*
 531          * Superusers bypass all permission-checking.
 532          */
 533         if (((Form_pg_shadow) GETSTRUCT(tuple))->usesuper)
 534         {
 535 #ifdef ACLDEBUG
 536                 elog(DEBUG, "pg_ownercheck: user \"%s\" is superuser",
 537                          usename);
 538 #endif
 539                 ReleaseSysCache(tuple);
 540                 return true;
 541         }
 542
 543         ReleaseSysCache(tuple);
 544         /* caution: usename is inaccessible beyond this point... */
 545
 546         tuple = SearchSysCache(cacheid,
 547                                                    PointerGetDatum(name),
 548                                                    0, 0, 0);
 549         switch (cacheid)
 550         {
 551                 case RELNAME:
 552                         if (!HeapTupleIsValid(tuple))
 553                                 elog(ERROR, "pg_ownercheck: class \"%s\" not found",
 554                                          name);
 555                         owner_id = ((Form_pg_class) GETSTRUCT(tuple))->relowner;
 556                         break;
 557                 case TYPENAME:
 558                         if (!HeapTupleIsValid(tuple))
 559                                 elog(ERROR, "pg_ownercheck: type \"%s\" not found",
 560                                          name);
 561                         owner_id = ((Form_pg_type) GETSTRUCT(tuple))->typowner;
 562                         break;
 563                 default:
 564                         elog(ERROR, "pg_ownercheck: invalid cache id: %d", cacheid);
 565                         owner_id = 0;           /* keep compiler quiet */
 566                         break;
 567         }
 568
 569         ReleaseSysCache(tuple);
 570
 571         return userid == owner_id;
 572 }
 573
 574 /*
 575  * Ownership check for an operator (specified by OID).
 576  */
 577 bool
 578 pg_oper_ownercheck(Oid userid, Oid oprid)
 579 {
 580         HeapTuple       tuple;
 581         AclId           owner_id;
 582         char       *usename;
 583
 584         tuple = SearchSysCache(SHADOWSYSID,
 585                                                    ObjectIdGetDatum(userid),
 586                                                    0, 0, 0);
 587         if (!HeapTupleIsValid(tuple))
 588                 elog(ERROR, "pg_oper_ownercheck: invalid user id %u",
 589                          (unsigned) userid);
 590         usename = NameStr(((Form_pg_shadow) GETSTRUCT(tuple))->usename);
 591
 592         /*
 593          * Superusers bypass all permission-checking.
 594          */
 595         if (((Form_pg_shadow) GETSTRUCT(tuple))->usesuper)
 596         {
 597 #ifdef ACLDEBUG
 598                 elog(DEBUG, "pg_ownercheck: user \"%s\" is superuser",
 599                          usename);
 600 #endif
 601                 ReleaseSysCache(tuple);
 602                 return true;
 603         }
 604
 605         ReleaseSysCache(tuple);
 606         /* caution: usename is inaccessible beyond this point... */
 607
 608         tuple = SearchSysCache(OPEROID,
 609                                                    ObjectIdGetDatum(oprid),
 610                                                    0, 0, 0);
 611         if (!HeapTupleIsValid(tuple))
 612                 elog(ERROR, "pg_ownercheck: operator %u not found",
 613                          oprid);
 614
 615         owner_id = ((Form_pg_operator) GETSTRUCT(tuple))->oprowner;
 616
 617         ReleaseSysCache(tuple);
 618
 619         return userid == owner_id;
 620 }
 621
 622 /*
 623  * Ownership check for a function (specified by name and argument types).
 624  */
 625 bool
 626 pg_func_ownercheck(Oid userid,
 627                                    char *funcname,
 628                                    int nargs,
 629                                    Oid *arglist)
 630 {
 631         HeapTuple       tuple;
 632         AclId           owner_id;
 633         char       *usename;
 634
 635         tuple = SearchSysCache(SHADOWSYSID,
 636                                                    ObjectIdGetDatum(userid),
 637                                                    0, 0, 0);
 638         if (!HeapTupleIsValid(tuple))
 639                 elog(ERROR, "pg_func_ownercheck: invalid user id %u",
 640                          (unsigned) userid);
 641         usename = NameStr(((Form_pg_shadow) GETSTRUCT(tuple))->usename);
 642
 643         /*
 644          * Superusers bypass all permission-checking.
 645          */
 646         if (((Form_pg_shadow) GETSTRUCT(tuple))->usesuper)
 647         {
 648 #ifdef ACLDEBUG
 649                 elog(DEBUG, "pg_ownercheck: user \"%s\" is superuser",
 650                          usename);
 651 #endif
 652                 ReleaseSysCache(tuple);
 653                 return true;
 654         }
 655
 656         ReleaseSysCache(tuple);
 657         /* caution: usename is inaccessible beyond this point... */
 658
 659         tuple = SearchSysCache(PROCNAME,
 660                                                    PointerGetDatum(funcname),
 661                                                    Int32GetDatum(nargs),
 662                                                    PointerGetDatum(arglist),
 663                                                    0);
 664         if (!HeapTupleIsValid(tuple))
 665                 func_error("pg_func_ownercheck", funcname, nargs, arglist, NULL);
 666
 667         owner_id = ((Form_pg_proc) GETSTRUCT(tuple))->proowner;
 668
 669         ReleaseSysCache(tuple);
 670
 671         return userid == owner_id;
 672 }
 673
 674 /*
 675  * Ownership check for an aggregate function (specified by name and
 676  * argument type).
 677  */
 678 bool
 679 pg_aggr_ownercheck(Oid userid,
 680                                    char *aggname,
 681                                    Oid basetypeID)
 682 {
 683         HeapTuple       tuple;
 684         AclId           owner_id;
 685         char       *usename;
 686
 687         tuple = SearchSysCache(SHADOWSYSID,
 688                                                    PointerGetDatum(userid),
 689                                                    0, 0, 0);
 690         if (!HeapTupleIsValid(tuple))
 691                 elog(ERROR, "pg_aggr_ownercheck: invalid user id %u",
 692                          (unsigned) userid);
 693         usename = NameStr(((Form_pg_shadow) GETSTRUCT(tuple))->usename);
 694
 695         /*
 696          * Superusers bypass all permission-checking.
 697          */
 698         if (((Form_pg_shadow) GETSTRUCT(tuple))->usesuper)
 699         {
 700 #ifdef ACLDEBUG
 701                 elog(DEBUG, "pg_aggr_ownercheck: user \"%s\" is superuser",
 702                          usename);
 703 #endif
 704                 ReleaseSysCache(tuple);
 705                 return true;
 706         }
 707
 708         ReleaseSysCache(tuple);
 709         /* caution: usename is inaccessible beyond this point... */
 710
 711         tuple = SearchSysCache(AGGNAME,
 712                                                    PointerGetDatum(aggname),
 713                                                    ObjectIdGetDatum(basetypeID),
 714                                                    0, 0);
 715         if (!HeapTupleIsValid(tuple))
 716                 agg_error("pg_aggr_ownercheck", aggname, basetypeID);
 717
 718         owner_id = ((Form_pg_aggregate) GETSTRUCT(tuple))->aggowner;
 719
 720         ReleaseSysCache(tuple);
 721
 722         return userid == owner_id;
 723 }