Postgresql RBAC

rbac_defaults.sql

DROP TYPE IF EXISTS "role" CASCADE;
DROP TYPE IF EXISTS "scope" CASCADE;
DROP TYPE IF EXISTS "action" CASCADE;
DROP TABLE IF EXISTS "roles" CASCADE;
DROP TABLE IF EXISTS "permissions" CASCADE;
DROP TABLE IF EXISTS "user_roles" CASCADE;

CREATE TYPE "role" as enum ('administrator', 'moderator');
CREATE TYPE "scope" as enum ('authentication', 'authorization');
CREATE TYPE "action" as enum ('read', 'write');

CREATE TABLE "roles" (
  "id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  "name" role NOT NULL
);
CREATE TABLE "permissions" (
  "id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  "role_id" bigint REFERENCES "roles" ON DELETE CASCADE NOT NULL,
  "scope" scope NOT NULL,
  "actions" action[] NOT NULL
);
CREATE TABLE "user_roles" (
  "id" bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  "user_id" uuid REFERENCES auth.users ON DELETE CASCADE NOT NULL,
  "role_id" bigint REFERENCES "roles" ON DELETE CASCADE NOT NULL
);

ALTER TABLE "roles" ENABLE ROW LEVEL SECURITY;
ALTER TABLE "permissions" ENABLE ROW LEVEL SECURITY;
ALTER TABLE "user_roles" ENABLE ROW LEVEL SECURITY;

CREATE OR REPLACE FUNCTION is_authorized (
  param_user_id uuid,
  param_permission_scope scope,
  param_permission_action action
)
RETURNS boolean
LANGUAGE plpgsql
as $$
declare
  result boolean;
begin
  SELECT 1 INTO result FROM user_roles
  WHERE user_roles.user_id = param_user_id
  AND EXISTS (
      SELECT 1 FROM permissions
      WHERE permissions.role_id = user_roles.role_id
      AND param_permission_scope = permissions.scope
      AND param_permission_action = ANY(permissions.actions)
  );
  result = COALESCE(result, false);
  return result;
end;
$$;

-- [x] roles SELECT
CREATE POLICY "roles: select" ON "roles" AS PERMISSIVE
FOR SELECT TO authenticated USING (
  EXISTS (
    SELECT 1 FROM user_roles
    WHERE user_roles.role_id = roles.id
    AND user_roles.user_id = auth.uid()
  )
);

-- [x] permissions SELECT
CREATE POLICY "permissions: select" ON "permissions" AS PERMISSIVE
FOR SELECT TO authenticated USING (
  EXISTS (
    SELECT 1 FROM roles
    WHERE roles.id = permissions.role_id
    AND EXISTS (
      SELECT 1 FROM user_roles
      WHERE user_roles.role_id = roles.id
      AND user_roles.user_id = auth.uid()
    )
  )
);

-- [x] user_roles SELECT
CREATE POLICY "user_roles: select" ON "user_roles" AS PERMISSIVE
FOR SELECT TO authenticated USING (
  user_roles.user_id = auth.uid()
);

-- [x] user_roles INSERT
CREATE POLICY "user_roles: insert" ON "user_roles" AS PERMISSIVE
FOR INSERT TO authenticated WITH CHECK (
  is_authorized(auth.uid(), 'authorization', 'write') = true
);

-- [x] user_roles DELETE
CREATE POLICY "user_roles: delete" ON "user_roles" AS PERMISSIVE
FOR DELETE TO authenticated USING (
  is_authorized(auth.uid(), 'authorization', 'write') = true
);

-- [x] administrator role
INSERT INTO "roles" ("name")
VALUES ('administrator');

-- [x] moderator role
INSERT INTO "roles" ("name")
VALUES ('moderator');

-- [x] administrator permissions
INSERT INTO "permissions" ("role_id", "scope", "actions")
VALUES (
  (SELECT "id" FROM "roles" WHERE "name" = 'administrator'),
  'authentication',
  ARRAY['read', 'write']::action[]
);
INSERT INTO "permissions" ("role_id", "scope", "actions")
VALUES (
  (SELECT "id" FROM "roles" WHERE "name" = 'administrator'),
  'authorization',
  ARRAY['read', 'write']::action[]
);

-- [x] moderator permissions
INSERT INTO "permissions" ("role_id", "scope", "actions")
VALUES (
  (SELECT "id" FROM "roles" WHERE "name" = 'moderator'),
  'authentication',
  ARRAY['read']::action[]
);
INSERT INTO "permissions" ("role_id", "scope", "actions")
VALUES (
  (SELECT "id" FROM "roles" WHERE "name" = 'moderator'),
  'authorization',
  ARRAY['read']::action[]
);

INSERT INTO "user_roles" ("user_id", "role_id")
VALUES (
  (SELECT "id" FROM "users" WHERE "email" = '[email protected]'),
  (SELECT "id" FROM "roles" WHERE "name" = 'administrator')
);

rbac_is_authorized.sql

SELECT
  "id",
  "email",
  is_authorized("id", 'authentication', 'read') as "authn_read",
  is_authorized("id", 'authentication', 'write') as "authn_write",
  is_authorized("id", 'authorization', 'read') as "authz_read",
  is_authorized("id", 'authorization', 'write') as "authz_write"
FROM "users";