joungmin/vpd-permission-poc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix #456: add VPD predicate rule tests

Browse Source
This commit is contained in:
devmrko
2026-06-25 21:21:48 +09:00
parent 5eaffd6229
commit d97dab34bd
5 changed files with 407 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
sql/adb/17_agent_ords_security_local_vpd_setup.sql
View File

@@ -341,22 +341,34 @@ BEGIN
RETURN '1 = 1';
ELSIF r.rule_type = 'MY_DEPT' THEN
v_column := safe_column(p_schema, p_object, NVL(r.rule_column, 'DEPT_CODE'));
append_or(v_column || ' = SYS_CONTEXT(''CB_AGENT_CTX'', ''DEPT_CODE'')');
IF v_column IS NOT NULL THEN
append_or(v_column || ' = SYS_CONTEXT(''CB_AGENT_CTX'', ''DEPT_CODE'')');
END IF;
ELSIF r.rule_type = 'SELF' THEN
v_column := safe_column(p_schema, p_object, NVL(r.rule_column, 'OWNER_EMP_NO'));
append_or(v_column || ' = SYS_CONTEXT(''CB_AGENT_CTX'', ''EMP_NO'')');
IF v_column IS NOT NULL THEN
append_or(v_column || ' = SYS_CONTEXT(''CB_AGENT_CTX'', ''EMP_NO'')');
END IF;
ELSIF r.rule_type = 'DEPT' THEN
v_column := safe_column(p_schema, p_object, NVL(r.rule_column, 'DEPT_CODE'));
append_or(v_column || ' = ' || quote_literal(r.rule_value));
IF v_column IS NOT NULL THEN
append_or(v_column || ' = ' || quote_literal(r.rule_value));
END IF;
ELSIF r.rule_type = 'EMP_NO' THEN
v_column := safe_column(p_schema, p_object, NVL(r.rule_column, 'OWNER_EMP_NO'));
append_or(v_column || ' = ' || quote_literal(r.rule_value));
IF v_column IS NOT NULL THEN
append_or(v_column || ' = ' || quote_literal(r.rule_value));
END IF;
ELSIF r.rule_type = '=' THEN
v_column := safe_column(p_schema, p_object, r.rule_column);
append_or('TO_CHAR(' || v_column || ') = ' || quote_literal(r.rule_value));
IF v_column IS NOT NULL THEN
append_or('TO_CHAR(' || v_column || ') = ' || quote_literal(r.rule_value));
END IF;
ELSIF r.rule_type IN ('!=', '<>') THEN
v_column := safe_column(p_schema, p_object, r.rule_column);
append_or('TO_CHAR(' || v_column || ') <> ' || quote_literal(r.rule_value));
IF v_column IS NOT NULL THEN
append_or('TO_CHAR(' || v_column || ') <> ' || quote_literal(r.rule_value));
END IF;
END IF;
END;
END LOOP;

28
sql/adb/26_agent_ords_security_dynamic_vpd_filter.sql
View File

@@ -18,6 +18,8 @@ CREATE OR REPLACE FUNCTION cb_agent_doc_vpd_filter(
) RETURN VARCHAR2
AUTHID DEFINER
AS
-- This function is intentionally whitelist-first:
-- no matched stored row rule means no rows are visible.
v_user_id NUMBER;
v_target VARCHAR2(128);
v_predicate VARCHAR2(32767);
@@ -104,23 +106,37 @@ BEGIN
IF r.rule_type = 'ALL' THEN
RETURN '1 = 1';
ELSIF r.rule_type = 'MY_DEPT' THEN
-- NULL rule_column keeps backward compatibility with the demo seed.
v_column := safe_column(p_schema, p_object, NVL(r.rule_column, 'DEPT_CODE'));
append_or(v_column || ' = SYS_CONTEXT(''CB_AGENT_CTX'', ''DEPT_CODE'')');
IF v_column IS NOT NULL THEN
append_or(v_column || ' = SYS_CONTEXT(''CB_AGENT_CTX'', ''DEPT_CODE'')');
END IF;
ELSIF r.rule_type = 'SELF' THEN
-- NULL rule_column keeps backward compatibility with the demo seed.
v_column := safe_column(p_schema, p_object, NVL(r.rule_column, 'OWNER_EMP_NO'));
append_or(v_column || ' = SYS_CONTEXT(''CB_AGENT_CTX'', ''EMP_NO'')');
IF v_column IS NOT NULL THEN
append_or(v_column || ' = SYS_CONTEXT(''CB_AGENT_CTX'', ''EMP_NO'')');
END IF;
ELSIF r.rule_type = 'DEPT' THEN
v_column := safe_column(p_schema, p_object, NVL(r.rule_column, 'DEPT_CODE'));
append_or(v_column || ' = ' || quote_literal(r.rule_value));
IF v_column IS NOT NULL THEN
append_or(v_column || ' = ' || quote_literal(r.rule_value));
END IF;
ELSIF r.rule_type = 'EMP_NO' THEN
v_column := safe_column(p_schema, p_object, NVL(r.rule_column, 'OWNER_EMP_NO'));
append_or(v_column || ' = ' || quote_literal(r.rule_value));
IF v_column IS NOT NULL THEN
append_or(v_column || ' = ' || quote_literal(r.rule_value));
END IF;
ELSIF r.rule_type = '=' THEN
v_column := safe_column(p_schema, p_object, r.rule_column);
append_or('TO_CHAR(' || v_column || ') = ' || quote_literal(r.rule_value));
IF v_column IS NOT NULL THEN
append_or('TO_CHAR(' || v_column || ') = ' || quote_literal(r.rule_value));
END IF;
ELSIF r.rule_type IN ('!=', '<>') THEN
v_column := safe_column(p_schema, p_object, r.rule_column);
append_or('TO_CHAR(' || v_column || ') <> ' || quote_literal(r.rule_value));
IF v_column IS NOT NULL THEN
append_or('TO_CHAR(' || v_column || ') <> ' || quote_literal(r.rule_value));
END IF;
END IF;
END;
END LOOP;

191
sql/adb/27_agent_ords_security_dynamic_vpd_filter_test.sql Normal file
View File

@@ -0,0 +1,191 @@
-- ============================================================
-- 27_agent_ords_security_dynamic_vpd_filter_test.sql
-- Unit-style checks for cb_agent_doc_vpd_filter.
--
-- Run after:
-- @sql/adb/26_agent_ords_security_dynamic_vpd_filter.sql
-- ============================================================
WHENEVER SQLERROR EXIT SQL.SQLCODE
SET ECHO OFF
SET FEEDBACK ON
SET DEFINE OFF
SET SERVEROUTPUT ON
PROMPT === Preparing dynamic VPD filter unit test data ===
BEGIN
cb_agent_ctx_pkg.clear_user;
DELETE FROM cb_permission_rule
WHERE perm_id BETWEEN 456000 AND 456999;
DELETE FROM cb_permission
WHERE perm_id BETWEEN 456000 AND 456999;
DELETE FROM cb_user_role
WHERE role_id BETWEEN 456000 AND 456999;
DELETE FROM cb_app_role
WHERE role_id BETWEEN 456000 AND 456999;
DELETE FROM cb_app_user
WHERE user_id BETWEEN 456000 AND 456999;
INSERT INTO cb_app_user(user_id, user_name, employee_no, dept_code, can_read_contents, active)
VALUES (456101, 'rule_test_hr', 'E456101', 'HR', 'N', 'Y');
INSERT INTO cb_app_user(user_id, user_name, employee_no, dept_code, can_read_contents, active)
VALUES (456102, 'rule_test_self', 'E2001', 'FIN', 'N', 'Y');
INSERT INTO cb_app_user(user_id, user_name, employee_no, dept_code, can_read_contents, active)
VALUES (456103, 'rule_test_all', 'E456103', 'HQ', 'Y', 'Y');
INSERT INTO cb_app_user(user_id, user_name, employee_no, dept_code, can_read_contents, active)
VALUES (456104, 'rule_test_eq', 'E456104', 'QA', 'N', 'Y');
INSERT INTO cb_app_user(user_id, user_name, employee_no, dept_code, can_read_contents, active)
VALUES (456105, 'rule_test_ne', 'E456105', 'QA', 'N', 'Y');
INSERT INTO cb_app_user(user_id, user_name, employee_no, dept_code, can_read_contents, active)
VALUES (456106, 'rule_test_bad_column', 'E456106', 'QA', 'N', 'Y');
INSERT INTO cb_app_role(role_id, role_name) VALUES (456101, 'RULE_TEST_MY_DEPT');
INSERT INTO cb_app_role(role_id, role_name) VALUES (456102, 'RULE_TEST_SELF');
INSERT INTO cb_app_role(role_id, role_name) VALUES (456103, 'RULE_TEST_ALL');
INSERT INTO cb_app_role(role_id, role_name) VALUES (456104, 'RULE_TEST_EQ');
INSERT INTO cb_app_role(role_id, role_name) VALUES (456105, 'RULE_TEST_NE');
INSERT INTO cb_app_role(role_id, role_name) VALUES (456106, 'RULE_TEST_BAD_COLUMN');
INSERT INTO cb_user_role(user_id, role_id) VALUES (456101, 456101);
INSERT INTO cb_user_role(user_id, role_id) VALUES (456102, 456102);
INSERT INTO cb_user_role(user_id, role_id) VALUES (456103, 456103);
INSERT INTO cb_user_role(user_id, role_id) VALUES (456104, 456104);
INSERT INTO cb_user_role(user_id, role_id) VALUES (456105, 456105);
INSERT INTO cb_user_role(user_id, role_id) VALUES (456106, 456106);
INSERT INTO cb_permission(perm_id, role_id, target_name, action_name)
VALUES (456101, 456101, 'CB_V_SEARCH_DOCUMENTS', 'SELECT');
INSERT INTO cb_permission(perm_id, role_id, target_name, action_name)
VALUES (456102, 456102, 'CB_V_SEARCH_DOCUMENTS', 'SELECT');
INSERT INTO cb_permission(perm_id, role_id, target_name, action_name)
VALUES (456103, 456103, 'CB_V_SEARCH_DOCUMENTS', 'SELECT');
INSERT INTO cb_permission(perm_id, role_id, target_name, action_name)
VALUES (456104, 456104, 'CB_V_SEARCH_DOCUMENTS', 'SELECT');
INSERT INTO cb_permission(perm_id, role_id, target_name, action_name)
VALUES (456105, 456105, 'CB_V_SEARCH_DOCUMENTS', 'SELECT');
INSERT INTO cb_permission(perm_id, role_id, target_name, action_name)
VALUES (456106, 456106, 'CB_V_SEARCH_DOCUMENTS', 'SELECT');
INSERT INTO cb_permission_rule(rule_id, perm_id, rule_column, rule_type, rule_value)
VALUES (456101, 456101, NULL, 'MY_DEPT', NULL);
INSERT INTO cb_permission_rule(rule_id, perm_id, rule_column, rule_type, rule_value)
VALUES (456102, 456102, NULL, 'SELF', NULL);
INSERT INTO cb_permission_rule(rule_id, perm_id, rule_column, rule_type, rule_value)
VALUES (456103, 456103, NULL, 'ALL', NULL);
INSERT INTO cb_permission_rule(rule_id, perm_id, rule_column, rule_type, rule_value)
VALUES (456104, 456104, 'DOC_ID', '=', '1');
INSERT INTO cb_permission_rule(rule_id, perm_id, rule_column, rule_type, rule_value)
VALUES (456105, 456105, 'DEPT_CODE', '!=', 'HR');
INSERT INTO cb_permission_rule(rule_id, perm_id, rule_column, rule_type, rule_value)
VALUES (456106, 456106, 'NO_SUCH_COLUMN', '=', 'x');
COMMIT;
END;
/
PROMPT === Running dynamic VPD filter unit checks ===
DECLARE
PROCEDURE fail(p_label IN VARCHAR2, p_expected IN VARCHAR2, p_actual IN VARCHAR2) IS
BEGIN
RAISE_APPLICATION_ERROR(
-20456,
p_label || ' expected [' || p_expected || '] actual [' || p_actual || ']'
);
END;
PROCEDURE assert_equals(
p_label IN VARCHAR2,
p_user_id IN NUMBER,
p_expected IN VARCHAR2
) IS
v_actual VARCHAR2(32767);
BEGIN
IF p_user_id IS NULL THEN
cb_agent_ctx_pkg.clear_user;
ELSE
cb_agent_ctx_pkg.set_user(p_user_id);
END IF;
v_actual := cb_agent_doc_vpd_filter('ADMIN', 'CB_V_SEARCH_DOCUMENTS');
IF v_actual != p_expected THEN
fail(p_label, p_expected, v_actual);
END IF;
DBMS_OUTPUT.PUT_LINE('PASS ' || p_label || ': ' || v_actual);
END;
PROCEDURE assert_contains(
p_label IN VARCHAR2,
p_user_id IN NUMBER,
p_expected IN VARCHAR2
) IS
v_actual VARCHAR2(32767);
BEGIN
cb_agent_ctx_pkg.set_user(p_user_id);
v_actual := cb_agent_doc_vpd_filter('ADMIN', 'CB_V_SEARCH_DOCUMENTS');
IF INSTR(v_actual, p_expected) = 0 THEN
fail(p_label, p_expected, v_actual);
END IF;
DBMS_OUTPUT.PUT_LINE('PASS ' || p_label || ': ' || v_actual);
END;
BEGIN
assert_equals('NO_CONTEXT_DENIES', NULL, '1 = 0');
assert_contains('MY_DEPT_DEFAULT_COLUMN', 456101, 'DEPT_CODE = SYS_CONTEXT(''CB_AGENT_CTX'', ''DEPT_CODE'')');
assert_contains('SELF_DEFAULT_COLUMN', 456102, 'OWNER_EMP_NO = SYS_CONTEXT(''CB_AGENT_CTX'', ''EMP_NO'')');
assert_equals('ALL_ALLOWS', 456103, '1 = 1');
assert_contains('EQUALS_RULE', 456104, 'TO_CHAR(DOC_ID) = ''1''');
assert_contains('NOT_EQUALS_RULE', 456105, 'TO_CHAR(DEPT_CODE) <> ''HR''');
assert_equals('BAD_COLUMN_DENIES', 456106, '1 = 0');
cb_agent_ctx_pkg.clear_user;
END;
/
PROMPT === Cleaning dynamic VPD filter unit test data ===
BEGIN
cb_agent_ctx_pkg.clear_user;
DELETE FROM cb_permission_rule
WHERE perm_id BETWEEN 456000 AND 456999;
DELETE FROM cb_permission
WHERE perm_id BETWEEN 456000 AND 456999;
DELETE FROM cb_user_role
WHERE role_id BETWEEN 456000 AND 456999;
DELETE FROM cb_app_role
WHERE role_id BETWEEN 456000 AND 456999;
DELETE FROM cb_app_user
WHERE user_id BETWEEN 456000 AND 456999;
COMMIT;
END;
/
PROMPT === Dynamic VPD filter unit checks passed ===