فیلتر مجوز اسناد پیش از بازیابی؛ آزمون سیاست دسترسی در پایتون

در جست‌وجوی یک پایگاه دانش مهندسی، مرتبط‌بودن متن با پرسش کافی نیست. ابتدا باید معلوم باشد کدام سند برای درخواست حاضر واجد شرایط است. این درس یک فیلتر کوچک روی فرادادهٔ کاملاً ساختگی می‌سازد و سپس فقط برای سندهای مجاز، تابع امتیازدهی محلی را صدا می‌زند. هدف، آزمایش ترتیب و منطق تصمیم است؛ هیچ حساب کاربری، مخزن محرمانه یا سرویس واقعی به برنامه متصل نیست.

۱. هویت، مجوز و ارتباط متن سه قرارداد جدا هستند

نام‌های LabA و LabB فضای دو سازمان فرضی‌اند. درخواست‌کنندهٔ نمونه به LabA تعلق دارد، عضو گروه power است و برچسب trained دارد. سندها نیز فضای سازمان، گروه‌های مجاز و مجوزهای لازم دارند. تمام این رشته‌ها را خودمان در دادهٔ آزمایش نوشته‌ایم؛ واردکردن نام یک گروه در دیکشنری، هویت یا عضویت واقعی شخص را اثبات نمی‌کند.

راهنمای مجوزدهی OWASP احراز هویت را از تصمیم مجوز جدا می‌کند و بر رد پیش‌فرض و کنترل هر درخواست تأکید دارد. این‌ها زمینهٔ طراحی تمرین‌اند؛ کد حاضر سامانهٔ احراز هویت یا مرز امنیتی آمادهٔ استقرار نیست. در کاربرد واقعی، منشأ قابل اعتماد ویژگی‌های درخواست‌کننده و محل اعمال مجوز نیز باید مشخص باشد.

می‌توان قرارداد جست‌وجو را در سه پرسش نوشت: چه کسی درخواست می‌کند؛ چه سندی برای او مجاز است؛ از میان سندهای مجاز، چه متنی با پرسش ارتباط دارد. در این تمرین پرسش اول ورودیِ فرض‌شده است، پرسش دوم پیاده‌سازی می‌شود و پرسش سوم با چند امتیاز دستی جایگزین شده است. بنابراین از خروجی، دقت بازیابی یا حفاظت واقعی اطلاعات نتیجه نمی‌گیریم.

۲. سیاست را با AND و OR بدون ابهام بنویسیم

مجوز نمونه سه شرط هم‌زمان دارد: فضای سازمان باید برابر باشد، دست‌کم یکی از گروه‌های درخواست‌کننده در فهرست گروه‌های مجاز سند باشد و همهٔ مجوزهای لازم سند در برچسب‌های او وجود داشته باشند. بین سه شرط AND داریم؛ داخل فهرست گروه‌های مجاز OR برقرار است؛ داخل مجوزهای لازم دوباره AND داریم. جایگزینی یکی از این عملگرها، سیاست دیگری می‌سازد.

سند D1 متعلق به LabA است، گروه power یا controls را می‌پذیرد و trained می‌خواهد. درخواست‌کننده با داشتن power و trained مجاز می‌شود و لازم نیست controls هم داشته باشد. سند D4 همان گروه power را می‌پذیرد، اما هم trained و هم approved می‌خواهد؛ داشتن فقط trained برای آن کافی نیست. برچسب‌ها، شرط‌های این مثال‌اند و صلاحیت واقعی مهندسی را ارزیابی نمی‌کنند.

فهرست خالی مجوزهای لازم یعنی شرط اضافی نداریم؛ هنوز برابری سازمان و اشتراک گروه لازم است. برعکس، گروه‌های مجاز نمی‌توانند خالی باشند. رشتهٔ admin هیچ امتیاز ضمنی ندارد و حرف بزرگ با کوچک برابر گرفته نمی‌شود. نقش تو‌در‌تو، وراثت گروه، استثنای عمومی و تقدم اجازه بر منع در این قرارداد تعریف نشده‌اند؛ نباید آن‌ها را از نام فیلدها حدس زد.

۳. فرادادهٔ ناقص را پیش از خواندن محتوا رد کنیم

طرح درخواست دقیقاً سه کلید tenant، groups و permits دارد. طرح سند دقیقاً doc_id، tenant، allow_groups و required_permits را می‌پذیرد. کلید کم، کلید ناشناخته یا نوع نامعتبر نتیجهٔ False می‌دهد. فهرست‌ها باید list واقعی با حداکثر هشت شناسهٔ یکتا باشند؛ رشتهٔ power به جای فهرست، خودکار تبدیل نمی‌شود. شناسه یک تا شانزده نویسهٔ ASCII دارد و با حرف آغاز می‌شود.

در شش ردیف دستی، D1 و D5 مجازند. D2 گروه دیگری دارد، D3 در سازمان دیگر است، D4 مجوز approved را کم دارد و D6 فاقد allow_groups است. تابع محلی فقط شناسه‌های D1 و D5 را برای خواندن امتیاز دریافت می‌کند. امتیازهای فرضی آن‌ها سه و یک‌اند؛ اعداد ۹۹، ۹۸، ۹۷ و ۹۶ برای ردیف‌های ردشده در آزمایش کنار گذاشته شده‌اند و از مسیر تابع خوانده نمی‌شوند.

اگر ابتدا همهٔ ردیف‌ها را رتبه‌بندی و فقط بهترین نتیجه را برداریم، سپس مجوز را بررسی کنیم، یک سند ردشده با امتیاز بالاتر می‌تواند جای نتیجهٔ مجاز را بگیرد. این کد ترتیب عکس را دارد: بررسی فراداده، خواندن امتیاز ردیف مجاز، مرتب‌سازی و انتخاب تعداد خواسته‌شده. این توضیح دربارهٔ همین مسیر محلی است؛ پیاده‌سازی فیلتر در موتور برداری یا پایگاه داده را آزمایش نکرده‌ایم.

۴. کد کامل و هجده بررسی با مرجع مستقل

فایل document_access_filter.py با Python 3.14.5 و کتابخانهٔ استاندارد اجرا شده است. مسیر اصلی از اشتراک و زیرمجموعهٔ مجموعه‌ها استفاده می‌کند. مرجع مستقل، روی داده‌های معتبر با حلقه‌های جدا گروه و مجوز را می‌گردد؛ ۳۸۴ ترکیب کوچک با دو فضای سازمان و حالت‌های مختلف ویژگی‌ها مقایسه می‌شوند. تمام آن ترکیب‌ها یک گروه بررسی‌اند و به شمار هجده گروه نام‌دار اضافه نمی‌شوند.

"""Synthetic metadata gate before a local scoring callback; not authentication.

All principals, documents and scores are fixtures. No cloud, network or real ACL.
"""
from copy import deepcopy
from itertools import product
import json
import platform
import re

IDENTIFIER = re.compile(r'[A-Za-z][A-Za-z0-9_-]{0,15}', re.ASCII)
SUBJECT_KEYS = {'tenant', 'groups', 'permits'}
DOCUMENT_KEYS = {'doc_id', 'tenant', 'allow_groups', 'required_permits'}


def valid_id(value):
    return type(value) is str and IDENTIFIER.fullmatch(value) is not None


def valid_labels(values, minimum=0):
    return (type(values) is list and minimum <= len(values) <= 8
            and all(valid_id(v) for v in values)
            and len(set(values)) == len(values))


def permitted(subject, document):
    """Fail closed on our exact synthetic schema, with no type coercion."""
    if type(subject) is not dict or set(subject) != SUBJECT_KEYS:
        return False
    if type(document) is not dict or set(document) != DOCUMENT_KEYS:
        return False
    if not (valid_id(subject['tenant']) and valid_labels(subject['groups'])
            and valid_labels(subject['permits']) and valid_id(document['doc_id'])
            and valid_id(document['tenant'])
            and valid_labels(document['allow_groups'], 1)
            and valid_labels(document['required_permits'])):
        return False
    return (subject['tenant'] == document['tenant']
            and bool(set(subject['groups']) & set(document['allow_groups']))
            and set(document['required_permits']) <= set(subject['permits']))


def rank_permitted(subject, documents, read_score, limit=2):
    """Gate each metadata row before invoking a local content-score callback."""
    if type(documents) is not list or len(documents) > 32:
        raise ValueError('documents')
    if type(limit) is not int or not 1 <= limit <= 16:
        raise ValueError('limit')
    if not callable(read_score):
        raise ValueError('read_score')
    # Reject ambiguous corpus IDs before any callback, including malformed rows.
    ids = [d['doc_id'] for d in documents
           if type(d) is dict and valid_id(d.get('doc_id'))]
    if len(set(ids)) != len(ids):
        raise ValueError('duplicate_doc_id')
    scored = []
    for document in documents:
        if not permitted(subject, document):
            continue
        score = read_score(document['doc_id'])
        if type(score) is not int or not 0 <= score <= 100:
            raise ValueError('score')
        scored.append({'doc_id': document['doc_id'], 'score': score})
    return sorted(scored, key=lambda row: (-row['score'], row['doc_id']))[:limit]


def document(doc_id='D1', tenant='LabA', groups=None, permits=None):
    return {'doc_id': doc_id, 'tenant': tenant,
            'allow_groups': ['power'] if groups is None else groups,
            'required_permits': [] if permits is None else permits}


def valid_domain_oracle(subject, metadata):
    """Independent loop formulation, used only on known-valid finite fixtures."""
    tenant_ok = subject['tenant'] == metadata['tenant']
    group_ok = False
    for allowed in metadata['allow_groups']:
        for member in subject['groups']:
            if allowed == member:
                group_ok = True
    requirements_ok = True
    for required in metadata['required_permits']:
        found = False
        for owned in subject['permits']:
            if required == owned:
                found = True
        if not found:
            requirements_ok = False
    return tenant_ok and group_ok and requirements_ok


def raises_value(call, message):
    try:
        call()
    except ValueError as error:
        return str(error) == message
    return False


def run_example():
    subject = {'tenant': 'LabA', 'groups': ['power'], 'permits': ['trained']}
    docs = [document('D1', groups=['power', 'controls'], permits=['trained']),
            document('D2', groups=['controls']),
            document('D3', tenant='LabB'),
            document('D4', permits=['trained', 'approved']),
            document('D5')]
    incomplete = document('D6')
    del incomplete['allow_groups']
    docs.append(incomplete)
    decisions = {d['doc_id']: permitted(subject, d) for d in docs}
    calls = []
    scores = {'D1': 3, 'D2': 99, 'D3': 98, 'D4': 97, 'D5': 1, 'D6': 96}

    def read_score(doc_id):
        calls.append(doc_id)
        if doc_id not in ('D1', 'D5'):
            raise RuntimeError('denied document reached callback')
        return scores[doc_id]

    ranked = rank_permitted(subject, docs, read_score)
    single = rank_permitted(subject, docs, lambda i: scores[i], 1)
    denied_calls = []
    denied_result = rank_permitted({}, docs, lambda i: denied_calls.append(i))
    before_subject, before_docs = deepcopy(subject), deepcopy(docs)
    rank_permitted(subject, docs, lambda i: scores[i])
    group_sets = [[], ['power'], ['controls'], ['power', 'controls']]
    permit_sets = [[], ['trained'], ['approved'], ['trained', 'approved']]
    oracle_cases = 0
    oracle_agreement = True
    for tenant, groups, permits, allowed, required in product(
            ['LabA', 'LabB'], group_sets, permit_sets, group_sets[1:], permit_sets):
        who = {'tenant': tenant, 'groups': groups, 'permits': permits}
        metadata = document(groups=allowed, permits=required)
        oracle_agreement &= permitted(who, metadata) == valid_domain_oracle(who, metadata)
        oracle_cases += 1
    missing_rows = []
    for key in DOCUMENT_KEYS:
        row = document()
        del row[key]
        missing_rows.append(row)
    unknown = document()
    unknown['public'] = True
    unknown_subject = dict(subject, admin=True)
    bad_documents = [document(groups='power'), document(groups=['power', 'power']),
                     document(groups=['*']), document(groups=['a'] * 9),
                     document(tenant=True), document(doc_id='D/1'),
                     document(permits=[None]), document(permits=('trained',))]
    bad_subjects = [None, [], {'tenant': 'LabA', 'permits': []},
                    dict(subject, groups='power'), dict(subject, permits=[True]),
                    dict(subject, groups=['power', 'power'])]
    tied = rank_permitted(subject, [docs[4], docs[0]], lambda i: 3)
    group_revoked = dict(subject, groups=['controls'])
    permission_revoked = deepcopy(docs[0])
    permission_revoked['required_permits'] = ['approved']
    duplicate_calls = []
    duplicate_rejected = raises_value(
        lambda: rank_permitted(subject, [docs[0], deepcopy(docs[0])],
                               lambda i: duplicate_calls.append(i)), 'duplicate_doc_id')
    checks = {
      'hand_six_metadata_decisions': decisions == {'D1': True, 'D2': False, 'D3': False,
                                                    'D4': False, 'D5': True, 'D6': False},
      'one_of_allowed_groups_is_sufficient': permitted(subject, docs[0]) is True,
      'tenant_is_required_even_when_group_matches': permitted(subject, docs[2]) is False,
      'every_required_permit_is_required': (not permitted(subject, docs[3])
          and permitted(dict(subject, permits=['trained', 'approved']), docs[3])),
      'empty_requirements_do_not_make_document_public': (permitted(subject, docs[4])
          and not permitted(dict(subject, groups=[]), docs[4])),
      'empty_allowed_groups_deny': not permitted(subject, document(groups=[])),
      'all_missing_document_fields_deny': all(not permitted(subject, d) for d in missing_rows),
      'unknown_subject_or_document_fields_deny': (not permitted(subject, unknown)
                                                   and not permitted(unknown_subject, docs[0])),
      'invalid_document_types_labels_and_bounds_deny': all(not permitted(subject, d) for d in bad_documents),
      'invalid_subject_shapes_and_types_deny': all(not permitted(w, docs[0]) for w in bad_subjects),
      'case_exact_and_no_implicit_admin_bypass': (not permitted(dict(subject, groups=['Power']), docs[4])
          and not permitted(dict(subject, groups=['admin']), docs[4])),
      'independent_valid_domain_loop_oracle': oracle_cases == 384 and bool(oracle_agreement),
      'only_permitted_rows_reach_callback': calls == ['D1', 'D5'] and ranked == [
          {'doc_id': 'D1', 'score': 3}, {'doc_id': 'D5', 'score': 1}],
      'denied_high_score_never_crowds_top_one': single == [{'doc_id': 'D1', 'score': 3}],
      'invalid_subject_reads_no_content': denied_calls == [] and denied_result == [],
      'fresh_subject_or_document_revocation_is_evaluated': (not permitted(group_revoked, docs[4])
          and not permitted(subject, permission_revoked)),
      'tie_order_and_inputs_remain_unchanged': (tied == [{'doc_id': 'D1', 'score': 3},
          {'doc_id': 'D5', 'score': 3}] and subject == before_subject and docs == before_docs),
      'ambiguous_corpus_and_invalid_pipeline_contracts_reject': (duplicate_rejected and duplicate_calls == []
          and all(raises_value(lambda x=x: rank_permitted(subject, docs, lambda i: 3, x), 'limit')
                  for x in [0, 17, True, 1.0])
          and raises_value(lambda: rank_permitted(subject, docs, lambda i: True), 'score')
          and raises_value(lambda: rank_permitted(subject, tuple(docs), lambda i: 3), 'documents')
          and raises_value(lambda: rank_permitted(subject, docs, None), 'read_score')),
    }
    assert len(checks) == 18 and all(type(v) is bool and v for v in checks.values()), checks
    return {'python': platform.python_version(), 'dependencies': 'Python standard library only',
            'scope': 'Synthetic exact metadata policy and sequential local callback; no real identity or security guarantee',
            'policy': 'same tenant AND any allowed group AND every required permit; invalid schema denies',
            'results': {'hand_decisions': decisions, 'content_callback_ids': calls, 'ranked': ranked,
                        'top_one': single, 'oracle_cases_inside_one_check': oracle_cases,
                        'denied_scores_never_read': {'D2': 99, 'D3': 98, 'D4': 97, 'D6': 96}},
            'checks': checks}


if __name__ == '__main__':
    print(json.dumps(run_example(), ensure_ascii=False, indent=2))

جواب دستی شش سند، کنترل فراخوانی تابع، رد گروه خالی، کلیدهای کم، نوع‌های غلط، حساسیت حروف و تغییر مجوز در فراخوانی بعدی جدا بررسی می‌شوند. در تساوی امتیاز، شناسهٔ سند ترتیب ثابت می‌دهد. تصویر ورودی پیش و پس از اجرا نیز مقایسه شده تا تغییر ناخواسته آشکار شود. درخواست نامعتبر، هیچ خواندن محتوایی در این مسیر ایجاد نمی‌کند.

{
  "python": "3.14.5",
  "dependencies": "Python standard library only",
  "scope": "Synthetic exact metadata policy and sequential local callback; no real identity or security guarantee",
  "policy": "same tenant AND any allowed group AND every required permit; invalid schema denies",
  "results": {
    "hand_decisions": {
      "D1": true,
      "D2": false,
      "D3": false,
      "D4": false,
      "D5": true,
      "D6": false
    },
    "content_callback_ids": [
      "D1",
      "D5"
    ],
    "ranked": [
      {
        "doc_id": "D1",
        "score": 3
      },
      {
        "doc_id": "D5",
        "score": 1
      }
    ],
    "top_one": [
      {
        "doc_id": "D1",
        "score": 3
      }
    ],
    "oracle_cases_inside_one_check": 384,
    "denied_scores_never_read": {
      "D2": 99,
      "D3": 98,
      "D4": 97,
      "D6": 96
    }
  },
  "checks": {
    "hand_six_metadata_decisions": true,
    "one_of_allowed_groups_is_sufficient": true,
    "tenant_is_required_even_when_group_matches": true,
    "every_required_permit_is_required": true,
    "empty_requirements_do_not_make_document_public": true,
    "empty_allowed_groups_deny": true,
    "all_missing_document_fields_deny": true,
    "unknown_subject_or_document_fields_deny": true,
    "invalid_document_types_labels_and_bounds_deny": true,
    "invalid_subject_shapes_and_types_deny": true,
    "case_exact_and_no_implicit_admin_bypass": true,
    "independent_valid_domain_loop_oracle": true,
    "only_permitted_rows_reach_callback": true,
    "denied_high_score_never_crowds_top_one": true,
    "invalid_subject_reads_no_content": true,
    "fresh_subject_or_document_revocation_is_evaluated": true,
    "tie_order_and_inputs_remain_unchanged": true,
    "ambiguous_corpus_and_invalid_pipeline_contracts_reject": true
  }
}

مجموعهٔ ورودی حداکثر سی‌ودو ردیف و حد انتخاب عدد صحیح یک تا شانزده است. شناسهٔ تکراری سند پیش از هر فراخوانی خطا می‌دهد تا یک شناسه به دو سیاست متناقض اشاره نکند. امتیاز باید عدد صحیح صفر تا صد باشد؛ bool امتیاز نیست. خطای تابع امتیازدهی نتیجهٔ رتبه‌بندی جزئی برنمی‌گرداند، اما برنامه برای متن خطای دلخواه آن تابع، پالایش امنیتی پیاده نکرده است.

۵. محدودیت کاربرد و چهار پرسش پاسخ‌دار

مستندات فیلتر رشته‌ای Microsoft تصریح می‌کند که مقایسهٔ شناسهٔ رشته‌ای، خودش احراز هویت نیست. مرور کنترل دسترسی سند نیز وابستگی تصمیم به فرادادهٔ همگام‌شده را توضیح می‌دهد. مثال ما فقط ورودی تازهٔ حافظه را می‌خواند؛ هماهنگی هویت، شاخص، قطعه‌های سند و ذخیره‌ساز، یا رقابت هم‌زمان را حل نمی‌کند.

پرسش اول: آیا power و controls هر دو برای D1 لازم‌اند؟ پاسخ: خیر؛ یک اشتراک کافی است. اما trained همچنان لازم است و فضای سازمان هم باید برابر بماند. مجازشدن از یک گروه، دو شرط دیگر را حذف نمی‌کند.

پرسش دوم: چرا نبود allow_groups به معنای سند عمومی نیست؟ پاسخ: چون حالت عمومی در این طرح نداریم. فیلد ناقص رد می‌شود؛ فهرست خالی هم اجازه نمی‌دهد. اضافه‌کردن public بدون تغییر رسمی طرح نیز کلید ناشناخته و موجب رد است.

پرسش سوم: آیا امتیاز ۹۹ می‌تواند D2 را وارد نتیجه کند؟ پاسخ: خیر؛ فیلتر، D2 را پیش از فراخوانی امتیازدهی حذف می‌کند. برای یادگیری مرحلهٔ تطبیق می‌توان نمایهٔ معکوس و جست‌وجوی بولی را خواند؛ قرارداد مجوز باید در ترکیب واقعی نیز مستقل بررسی شود.

پرسش چهارم: آیا قبولی آزمون‌ها برای استقرار RAG کافی است؟ پاسخ: خیر؛ حتی فرادادهٔ همهٔ سندها در حافظهٔ همین آزمایش حاضر است. برای اتصال به پایگاه دانش RAG باید مرز اعتماد، همگام‌سازی و مسیرهای دیگر ارزیابی شوند. منابع در ۲۴ مهر ۱۴۰۵ خوانده شدند؛ تاریخ به‌روزرسانی دو صفحهٔ Microsoft، ۲۴ اوت و ۱۷ سپتامبر ۲۰۲۶ است و این مطلب خبر تازه نیست.

منابع و مطالعهٔ بیشتر

• OWASP؛ مجوزدهی، رد پیش‌فرض و بررسی هر درخواست
• Microsoft؛ الگوی فیلتر رشته‌ای نتایج جست‌وجو
• Microsoft؛ کنترل دسترسی سند و همگام‌سازی فراداده

بررسی منابع: ۲۴ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.